# !/usr/bin/python # -*- coding: UTF-8 -*- from sqlalchemy import * from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import Session from sqlalchemy.ext.declarative import declarative_base # 创建对象的基类: Base = declarative_base() class PTable(Base): __tablename__ = 'tables' table_catalog = Column(String(512)) # table_schema = Column(String(64)) # table_name = Column(String(64), primary_key=True) table_type = Column(String(64)) # engine = Column(String(64)) # version = Column(BigInteger) # row_format = Column(String(10)) # table_rows = Column(BigInteger) # avg_row_length = Column(BigInteger) # data_length = Column(BigInteger) # max_data_length = Column(BigInteger) # index_length = Column(BigInteger) # data_free = Column(BigInteger) # auto_increment = Column(BigInteger) # create_time = Column(Date) # update_time = Column(Date) # check_time = Column(Date) # table_collation = Column(String(32)) # checksum = Column(BigInteger) # create_options = Column(String(2048)) # table_comment = Column(String(2048)) # class PColumn(Base): __tablename__ = 'columns' table_schema = Column(String(255)) table_name = Column(String(255)) column_name = Column(String(255), primary_key=True) data_type = Column(String(255)) character_maximum_length = Column(String(255)) #字符类型时，字段长度 column_key = Column(String(255)) #PRI为主键，UNI为unique，MUL是什么意思？ column_comment = Column(String(255)) #字段说明 extra = Column(String(255)) #'auto_increment' dic = {} dic["tinyint"] = "Boolean" dic["smallint"] = "SmallInteger" dic["mediumint"] = "Integer" dic["int"] = "Integer" dic["integer"] = "Integer" dic["bigint"] = "BigInteger" dic["float"] = "Float" dic["double"] = "Numeric" dic["decimal"] = "Numeric" dic["date"] = "Date" dic["time"] = "Time" dic["year"] = "Integer" dic["datetime"] = "Date" dic["timestamp"] = "Date" dic["char"] = "String" dic["varchar"] = "String" dic["tinyblob"] = "String" dic["tinytext"] = "String" dic["blob"] = "String" dic["text"] = "String" dic["mediumblob"] = "LargeBinary" dic["mediumtext"] = "String" dic["longblob"] = "LargeBinary" dic["longtext"] = "String" session = None; def open(host,port,db,user,pwd) : url = 'mysql+mysqlconnector://%s:%s@%s:%d/%s' % (user,pwd,host,port,db) engine = create_engine(url) DbSession = sessionmaker(bind=engine) session = DbSession() return session def close() : session.close() def generate_db(db_name) : tq = session.query(PTable) ts = tq.filter(PTable.table_schema==db_name).all() for t in ts: generate_table(t.table_name,t.table_comment) def generate_table(table_name,memoto) : print "" print "#%s" % memoto print "class %s(Base) : " % table_name print "" print " __tablename__ = '%s'" % table_name print "" cq = session.query(PColumn) cs = cq.filter(PColumn.table_name== table_name ).all() for c in cs: item = generate_column(c) print item def generate_column(c) : item = " " + c.column_name.lower()+" = Column("+dic[c.data_type] if c.character_maximum_length != None : item = item + "(%d)" % c.character_maximum_length if c.column_key == "PRI": item = item + ",primary_key=True" item = item + ")" item = item.ljust(60) if c.column_comment != None : item = item +"# "+c.column_comment return item session = open("127.0.0.1",3306,"information_schema","root","xxx") generate_db("netsharp") close()
生成的结果如下
# class sys_job(Base) : __tablename__ = 'sys_job' group_name = Column(String(50)) # cron = Column(String(50)) # cron_description = Column(String(500)) # description = Column(String(500)) # java_type = Column(String(500)) # par = Column(String(200)) # status = Column(Integer) # resource_node_id = Column(Integer) # code = Column(String(50)) # name = Column(String(200)) # memoto = Column(String(1000)) # id = Column(Integer,primary_key=True) # creator_id = Column(Integer) # creator = Column(String(50)) # create_time = Column(Date) # updator_id = Column(Integer) # updator = Column(String(50)) # update_time = Column(Date) # ts = Column(Date) # job_group = Column(String(50)) # cron_expression = Column(String(50)) # # class sys_job_log(Base) : __tablename__ = 'sys_job_log' group_name = Column(String(50)) # java_type = Column(String(500)) # operation = Column(String(50)) # timed = Column(BigInteger) # code = Column(String(50)) # name = Column(String(200)) # memoto = Column(String(1000)) # id = Column(Integer,primary_key=True) # creator_id = Column(Integer) # creator = Column(String(50)) # create_time = Column(Date) # updator_id = Column(Integer) # updator = Column(String(50)) # update_time = Column(Date) # ts = Column(Date) # job_group = Column(String(50)) # operation_date = Column(Date) #