python 从数据库表生成model,pythonmodel

来源:http://www.sh-fengwen.com 作者:鲜果干果 人气:58 发布时间:2019-09-03
摘要:python 从数据库表生成model,pythonmodel python 从数据库表生成model   找了很久才找到这个,我是新手... 现在已有建好的数据库,需要基于原有数据做数据分析的web应用,我选择python+Tornad

python 从数据库表生成model,pythonmodel

python 从数据库表生成model

 

找了很久才找到这个,我是新手...

现在已有建好的数据库,需要基于原有数据做数据分析的web应用,我选择python+Tornado ,由于不想写SQL语句,就想偷个懒

 

1、安装工具

1 [email protected]:~/code/py/django/logcloud$ sudo pip install sqlacodegen
2 Downloading/unpacking sqlacodegen
3   Downloading sqlacodegen-1.1.6-py2.py3-none-any.whl
4 Downloading/unpacking inflect>=0.2.0 (from sqlacodegen)
5   Downloading inflect-0.2.5-py2.py3-none-any.whl (58kB): 58kB downloaded
6 Requirement already satisfied (use --upgrade to upgrade): SQLAlchemy>=0.6.0 in /usr/local/lib/python2.7/dist-packages (from sqlacodegen)
7 Installing collected packages: sqlacodegen, inflect
8 Successfully installed sqlacodegen inflect
9 Cleaning up...

 

2、转换

[email protected]:~/code/py/django/logcloud$ sqlacodegen mssql+pymssql://name:[email protected]/LogColudDB --outfile logcloude_model.py
[email protected]:~/code/py/django/logcloud$

 

3、查看,哈哈

[email protected]:~/code/py/django/logcloud$ cat logcloude_model.py
# coding: utf-8
from sqlalchemy import BigInteger, Column, DateTime, Float, ForeignKey, Integer, LargeBinary, T                   able, Unicode, text
from sqlalchemy.dialects.mssql.base import BIT
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()
metadata = Base.metadata


class AuthorityInfo(Base):
    __tablename__ = 'AuthorityInfo'

    ID = Column(BigInteger, primary_key=True)
    Description = Column(Unicode('max'), nullable=False)
    IsDelete = Column(BIT, nullable=False)
    AuthorityIndex = Column(Integer, nullable=False)
    AuthorityName = Column(Unicode('max'), nullable=False)


class CPKInfoHistory(Base):
    __tablename__ = 'CPKInfoHistory'

    PO = Column(Unicode(10), primary_key=True)
    ProcessName = Column(Unicode(50), nullable=False)
    Result = Column(Unicode('max'), nullable=False)
    LastLogID = Column(BigInteger, nullable=False)


class ComputerState(Base):
    __tablename__ = 'ComputerState'

    Name = Column(Unicode(50), primary_key=True)
    Ip = Column(Unicode(50), nullable=False)
    IsDelete = Column(BIT, nullable=False)
    LastDate = Column(DateTime, nullable=False)
    IsProceted = Column(BIT, nullable=False)


class DeviceInfo(Base):
    __tablename__ = 'DeviceInfo'

    ID = Column(BigInteger, primary_key=True)
    ClientName = Column(Unicode(50), nullable=False)
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)
    LinesInfo_ID = Column(ForeignKey(u'LinesInfo.ID'), nullable=False, index=True)

    LinesInfo = relationship(u'LinesInfo')


class DisposeErrorCodeInfo(Base):
    __tablename__ = 'DisposeErrorCodeInfo'

    ID = Column(BigInteger, primary_key=True)
    NewOperation = Column(Unicode('max'), nullable=False)
    Status = Column(Unicode('max'), nullable=False)
    DisposeInfo_ID = Column(ForeignKey(u'DisposeInfo.ID'), nullable=False, index=True)
    ErrorCode_Info_ID = Column(ForeignKey(u'ErrorCode_Info.ID'), nullable=False, index=True)
    ErrorCode_OperationID = Column(BigInteger, nullable=False)
    WeightValue = Column(Integer, nullable=False)

    DisposeInfo = relationship(u'DisposeInfo')
    ErrorCode_Info = relationship(u'ErrorCodeInfo')


class DisposeErrorCodeInfoHistory(Base):
    __tablename__ = 'DisposeErrorCodeInfoHistory'

    ID = Column(BigInteger, primary_key=True)
    OperateTime = Column(DateTime, nullable=False)
    OldStatus = Column(Unicode('max'), nullable=False)
    NewStatus = Column(Unicode('max'), nullable=False)
    IsDelete = Column(BIT, nullable=False)
    WeightValue = Column(Integer, nullable=False)
    UserInfo_ID = Column(ForeignKey(u'UserInfo.ID'), nullable=False, index=True)
    DisposeErrorCodeInfo_ID = Column(ForeignKey(u'DisposeErrorCodeInfo.ID'), nullable=False, in                   dex=True)

    DisposeErrorCodeInfo = relationship(u'DisposeErrorCodeInfo')
    UserInfo = relationship(u'UserInfo')


class DisposeInfo(Base):
    __tablename__ = 'DisposeInfo'

    ID = Column(BigInteger, primary_key=True)
    Operation = Column(Unicode('max'), nullable=False)
    IsOK = Column(BIT, nullable=False)
    Description = Column(Unicode(1024))
    TouchTime = Column(DateTime, nullable=False)
    NoticeInfo_ID = Column(ForeignKey(u'NoticeInfo.ID'), nullable=False, index=True)
    UserTask_ID = Column(ForeignKey(u'UserTask.ID'), nullable=False, index=True)

    NoticeInfo = relationship(u'NoticeInfo')
    UserTask = relationship(u'UserTask')


class ErrorCodeInfo(Base):
    __tablename__ = 'ErrorCode_Info'

    ID = Column(BigInteger, primary_key=True)
    ErrorCode = Column(Unicode(10), nullable=False)
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)


class ErrorCodeOperation(Base):
    __tablename__ = 'ErrorCode_Operation'

    ID = Column(BigInteger, primary_key=True)
    Operation = Column(Unicode('max'), nullable=False)
    WeightValue = Column(Integer, nullable=False)
    IsEnable = Column(BIT, nullable=False)
    ErrorCode_Info_ID = Column(ForeignKey(u'ErrorCode_Info.ID'), nullable=False, index=True)

    ErrorCode_Info = relationship(u'ErrorCodeInfo')


class FilesManage(Base):
    __tablename__ = 'FilesManage'

    ID = Column(BigInteger, primary_key=True)
    ClassName = Column(Unicode(50), nullable=False, index=True)
    Md5 = Column(Unicode(32), nullable=False)
    Data = Column(LargeBinary, nullable=False)
    Ver = Column(Integer, nullable=False)
    DateCreated = Column(Unicode(50), nullable=False)
    UpLoadUserName = Column(Unicode(50), nullable=False)
    Remarks = Column(Unicode('max'))
    Catagory = Column(Unicode(50), nullable=False)
    LocalFileName = Column(Unicode(50))


class LOGInfo(Base):
    __tablename__ = 'LOG_Info'

    ID = Column(BigInteger, primary_key=True)
    Po = Column(Unicode(10), nullable=False, index=True)
    ProcessName = Column(Unicode(10), nullable=False, index=True)
    User = Column(Unicode(10), nullable=False)
    ErrorCode = Column(Unicode(10))
    Log = Column(Unicode('max'), nullable=False)
    Barcode = Column(Unicode(50))
    Isn = Column(Unicode(50))
    Shift = Column(Unicode(10), nullable=False)
    TestResult = Column(Unicode(10), nullable=False, index=True)
    LastDate = Column(DateTime, nullable=False)
    ClientName = Column(Unicode(50), nullable=False, index=True)


class LinesInfo(Base):
    __tablename__ = 'LinesInfo'

    ID = Column(BigInteger, primary_key=True)
    Name = Column(Unicode(50), nullable=False)
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)


class NoticeInfo(Base):
    __tablename__ = 'NoticeInfo'

    ID = Column(BigInteger, primary_key=True)
    Top1_ErrorCodeID = Column(BigInteger, nullable=False)
    Top2_ErrorCodeID = Column(BigInteger)
    Top3_ErrorCodeID = Column(BigInteger)
    TouchTime = Column(DateTime, nullable=False)
    IsDispose = Column(BIT, nullable=False)
    TaskLavel = Column(Integer, nullable=False)
    Responsibility_ID = Column(ForeignKey(u'Responsibility.ID'), nullable=False, index=True)
    UserTask_ID = Column(ForeignKey(u'UserTask.ID'), nullable=False, index=True)
    DeviceInfo_ID = Column(ForeignKey(u'DeviceInfo.ID'), nullable=False, index=True)
    StatisticsInfo_ID = Column(BigInteger, nullable=False)

    DeviceInfo = relationship(u'DeviceInfo')
    Responsibility = relationship(u'Responsibility')
    UserTask = relationship(u'UserTask')


class POInfo(Base):
    __tablename__ = 'POInfo'

    ID = Column(BigInteger, primary_key=True)
    Po = Column(Unicode(10), nullable=False)
    Plm = Column(Unicode(20))
    ProductName = Column(Unicode(50))
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)
    Customer = Column(Unicode(50))


class Responsibility(Base):
    __tablename__ = 'Responsibility'

    ID = Column(BigInteger, primary_key=True)
    ProcessName = Column(Unicode(10), nullable=False)
    BaseNumber = Column(Integer, nullable=False)
    ErrorRate = Column(Float(53), nullable=False)
    Description = Column(Unicode(1024))
    OverTime = Column(Integer, nullable=False)
    POInfo_ID = Column(ForeignKey(u'POInfo.ID'), nullable=False, index=True)
    UserInfo_ID = Column(ForeignKey(u'UserInfo.ID'), nullable=False, index=True)

    POInfo = relationship(u'POInfo')
    UserInfo = relationship(u'UserInfo')


class RoleAuthority(Base):
    __tablename__ = 'RoleAuthority'

    ID = Column(BigInteger, primary_key=True)
    IsDelete = Column(BIT, nullable=False)
    RoleInfo_ID = Column(ForeignKey(u'RoleInfo.ID'), nullable=False, index=True)
    AuthorityInfo_ID = Column(ForeignKey(u'AuthorityInfo.ID'), nullable=False, index=True)

    AuthorityInfo = relationship(u'AuthorityInfo')
    RoleInfo = relationship(u'RoleInfo')


class RoleInfo(Base):
    __tablename__ = 'RoleInfo'

    ID = Column(BigInteger, primary_key=True)
    RoleName = Column(Unicode(256), nullable=False)
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)
    RoleLevel = Column(Integer, nullable=False)


class SettingInfo(Base):
    __tablename__ = 'SettingInfo'

    ID = Column(BigInteger, primary_key=True, nullable=False)
    Key = Column(Unicode(50), primary_key=True, nullable=False)
    Value = Column(Unicode('max'), nullable=False)
    Description = Column(Unicode(1024))


class StatisticsInfo(Base):
    __tablename__ = 'StatisticsInfo'

    ID = Column(BigInteger, primary_key=True)
    ProcessName = Column(Unicode(10), nullable=False)
    BeginTime = Column(DateTime, nullable=False)
    NowErrorRate = Column(Float(53), nullable=False)
    Times = Column(Integer, nullable=False)
    IsOutmoded = Column(BIT, nullable=False)
    POInfo_ID = Column(ForeignKey(u'POInfo.ID'), nullable=False, index=True)
    DeviceInfo_ID = Column(ForeignKey(u'DeviceInfo.ID'), nullable=False, index=True)

    DeviceInfo = relationship(u'DeviceInfo')
    POInfo = relationship(u'POInfo')


class UserInfo(Base):
    __tablename__ = 'UserInfo'

    ID = Column(BigInteger, primary_key=True)
    Name = Column(Unicode(10), nullable=False)
    JobNumber = Column(Unicode(10), nullable=False)
    Phone = Column(Unicode(20))
    Emil = Column(Unicode(30), nullable=False)
    Department = Column(Unicode(20))
    Duties = Column(Unicode(20))
    Description = Column(Unicode(1024))
    Group = Column(Unicode(20), nullable=False)
    IsDelete = Column(BIT, nullable=False)
    Password = Column(Unicode(32))
    CreateDateTime = Column(DateTime, nullable=False)


class UserProfile(Base):
    __tablename__ = 'UserProfile'

    UserId = Column(Integer, primary_key=True)
    UserName = Column(Unicode(56), nullable=False, unique=True)


class UserRoleInfo(Base):
    __tablename__ = 'UserRoleInfo'

    ID = Column(BigInteger, primary_key=True)
    IsDelete = Column(BIT, nullable=False)
    UserInfo_ID = Column(ForeignKey(u'UserInfo.ID'), nullable=False, index=True)
    RoleInfo_ID = Column(ForeignKey(u'RoleInfo.ID'), nullable=False, index=True)

    RoleInfo = relationship(u'RoleInfo')
    UserInfo = relationship(u'UserInfo')


class UserTask(Base):
    __tablename__ = 'UserTask'

    ID = Column(BigInteger, primary_key=True)
    TaskLavel = Column(Integer, nullable=False)
    Shift = Column(Unicode(10), nullable=False)
    Description = Column(Unicode(1024))
    IsDelete = Column(BIT, nullable=False)
    UserInfo_ID = Column(ForeignKey(u'UserInfo.ID'), nullable=False, index=True)
    LinesInfo_ID = Column(ForeignKey(u'LinesInfo.ID'), nullable=False, index=True)

    LinesInfo = relationship(u'LinesInfo')
    UserInfo = relationship(u'UserInfo')


t_view_LinesState = Table(
    'view_LinesState', metadata,
    Column('ID', BigInteger, nullable=False),
    Column('ClientName', Unicode(50), nullable=False),
    Column('LinesInfo_ID', BigInteger, nullable=False),
    Column('count', Integer)
)


t_view_LogState = Table(
    'view_LogState', metadata,
    Column('Po', Unicode(10), nullable=False),
    Column('ProcessName', Unicode(10), nullable=False),
    Column('ErrorCode', Unicode(10)),
    Column('ClientName', Unicode(50), nullable=False),
    Column('ID', BigInteger),
    Column('count', Integer)
)


t_view_NoticeInfo = Table(
    'view_NoticeInfo', metadata,
    Column('ID', BigInteger, nullable=False),
    Column('TouchTime', DateTime, nullable=False),
    Column('Top1_ErrorCodeID', BigInteger, nullable=False),
    Column('Top2_ErrorCodeID', BigInteger),
    Column('Top3_ErrorCodeID', BigInteger),
    Column('IsDispose', BIT, nullable=False),
    Column('TaskLavel', Integer, nullable=False),
    Column('ClientName', Unicode(50)),
    Column('ProcessName', Unicode(10)),
    Column('BeginTime', DateTime),
    Column('NowErrorRate', Float(53)),
    Column('Times', Integer),
    Column('IsOutmoded', BIT),
    Column('Po', Unicode(10))
)


t_view_Top1_error = Table(
    'view_Top1_error', metadata,
    Column('Top1_ErrorCodeID', BigInteger, nullable=False),
    Column('count', Integer),
    Column('ID', BigInteger, nullable=False),
    Column('ErrorCode', Unicode(10), nullable=False),
    Column('Description', Unicode(1024))
)


class WebpagesMembership(Base):
    __tablename__ = 'webpages_Membership'

    UserId = Column(Integer, primary_key=True)
    CreateDate = Column(DateTime)
    ConfirmationToken = Column(Unicode(128))
    IsConfirmed = Column(BIT, server_default=text("((0))"))
    LastPasswordFailureDate = Column(DateTime)
    PasswordFailuresSinceLastSuccess = Column(Integer, nullable=False, server_default=text("((0                   ))"))
    Password = Column(Unicode(128), nullable=False)
    PasswordChangedDate = Column(DateTime)
    PasswordSalt = Column(Unicode(128), nullable=False)
    PasswordVerificationToken = Column(Unicode(128))
    PasswordVerificationTokenExpirationDate = Column(DateTime)


class WebpagesOAuthMembership(Base):
    __tablename__ = 'webpages_OAuthMembership'

    Provider = Column(Unicode(30), primary_key=True, nullable=False)
    ProviderUserId = Column(Unicode(100), primary_key=True, nullable=False)
    UserId = Column(Integer, nullable=False)


class WebpagesRole(Base):
    __tablename__ = 'webpages_Roles'

    RoleId = Column(Integer, primary_key=True)
    RoleName = Column(Unicode(256), nullable=False, unique=True)

    UserProfile = relationship(u'UserProfile', secondary='webpages_UsersInRoles')


t_webpages_UsersInRoles = Table(
    'webpages_UsersInRoles', metadata,
    Column('UserId', ForeignKey(u'UserProfile.UserId'), primary_key=True, nullable=False),
    Column('RoleId', ForeignKey(u'webpages_Roles.RoleId'), primary_key=True, nullable=False)
)
[email protected]:~/code/py/django/logcloud$

 

从数据库表生成model,pythonmodel python 从数据库表生成model 找了很久才找到这个,我是新手... 现在已有建好的数据库,需要基于原有数...

 

       在数据库设计完成之后, 常常需要在 wiki 或其他文档中保存一份数据库中所有表的 desc 描述, 尤其是每个字段的含义和用途。 手动去生成自然是不可取的。 因此, 我编写了一个简单的 python 程序,可以自动生成数据库中所有表的 desc 描述, 并以可读格式输出。

       

# -*- coding: utf-8 -*-
# -------------------------------------------------------------------------------
# Name:          db_tables_descs.py
# Purpose:       generate the tables that describe the meanings of fields in db
#
# Author:       qin.shuq
#
# Created:      2014/11/17
# Output:       desc.txt
#               recording the tables that describe the meanings of fields in db
#-------------------------------------------------------------------------------
#!/usr/bin/env python

import db

globalFieldDescs = ('Field', 'Type', 'Null', 'Key', 'Default', 'Extra')

globalDescFile = 'desc.txt'

conflictedWithMysqlKeywords = set(['group'])

fieldDescMapping = {
    'id':         '唯一标识',
    'is_deleted': '是否逻辑删除',
    'status':     '实体状态',
    'type':       '实体类型',
    'priority':   '优先级',
    'password':   '密码',
    'ip':         'ip 地址',
    'mac':        'mac 地址',
    'protocol':   '访问协议',
    'user_id':    '用户唯一标识'
}

def formatCols(fieldDesc):
    return  "%-16s %-24s %-5s %-8s %-8s %-30s" % fieldDesc

def withNewLine(astr):
    return astr + 'n'


def commonFieldsProcess(fieldDescList):
    fieldName = fieldDescList[0]
    fieldDesc = fieldDescMapping.get(fieldName)
    desclen =   len(fieldDescList)
    if fieldDesc is None:
        if fieldName.startswith('gmt_c'):
            fieldDesc = '创建时间'
        elif fieldName.startswith('gmt_m'):
            fieldDesc = '修改时间'
        else:
            fieldDesc = fieldDescList[desclen-1]
    fieldDescList[desclen-1] = fieldDesc

def formatF(fieldDescTuple):
    fieldDescList = list(fieldDescTuple)
    fieldLen = len(fieldDescList)
    for i in range(fieldLen):
        if fieldDescList[i] is None:
            fieldDescList[i] = 'NULL'
        else:
            fieldDescList[i] = str(fieldDescList[i])
    commonFieldsProcess(fieldDescList)
    return formatCols(tuple(fieldDescList))

def format(tableDesc):
    desc = ''
    for fieldDescTuple in tableDesc:
        desc += withNewLine(formatF(fieldDescTuple))
    return desc

def descDb(givenDb):
    tablesRet = givenDb.query("show tables;")
    tableNames = [table[0] for table in tablesRet]
    desc = u''
    for tablename in tableNames:
        if tablename in conflictedWithMysqlKeywords:
            tablename = '`' + tablename + '`'
        descSql = "desc " + tablename
        tableDesc = givenDb.query(descSql)
        desc += withNewLine(tablename)
        desc += withNewLine(formatCols(globalFieldDescs)).decode('utf-8')
        desc += withNewLine(format(tableDesc)).decode('utf-8')
        desc += withNewLine('').decode('utf-8')
    return desc


def main():

    descFile = open(globalDescFile, 'w')

    desc = descDb(db.Mydb())
    descFile.write(desc.encode('utf-8'))

    descFile.close()


if __name__ == '__main__':
    main()

db.py

#!/usr/ali/bin/python
# coding=utf-8

'''Implements a database api to your db.

Example 1: Query SQL

a. Use execute() method to execute query sql:

    db.execute('select * from ip')

    # Get only the first two rows
    db.get_rows(2)
    # result like [('10.10.0.1', 'my'), ..]

    # Get the next two rows, but each row record is a dict
    db.get_rows(2, is_dict = True)
    # result like [{'address':'10.10.0.1', 'name': 'my'}, ..]

b. Use query() method to execute query sql directly:

    # The query() method will get the result rows immediately
    db.query('select * from ip', size = 2, is_dict = True)

c. Use split_query() method to split long query into small ones:

   # Assume that the name_list's length is 10000
   # See the docstring of split_query() for more details
   db.split_query('select address from ip', 'name', name_list)

Example 2: Insert SQL

a. Insert a new record into ip table:

    db.execute("insert into ip('address','name') values('192.168.0.1','vm-xxx')")

    # If auto commit set to false, call commit() method manually
    db.commit()

b. Insert multi-records into ip table:

    db.executemany("insert into ip('address','name') values(%s,%s)", [
            ('192.168.0.1', 'vm-xxx'),
            ('192.168.0.2', 'vm-yyy'),
            ('192.168.0.3', 'vm-zzz')])
    db.commit()

Note: db.multi_insert is an alias for executemany method.
See test_main() method for more examples.
'''

from database import DB

class Mydb(DB):
    '''A simple query interface of a specific database.'''
    def __init__(self, read_only = True,
                 auto_commit = False, timeout = 5, auto_connect = False,
                 max_idle_time = 28800):
        '''Initialize the database access object.'''
        # Get the database parameters
        args = {'host':'127.0.0.1', 'user':'root','passwd':'123456','db':'mysql','port':3306,'charset':'utf8'}

        # Set extra connection parameters
        args['connect_timeout'] = timeout
        args['auto_commit'] = auto_commit
        args['max_idle_time'] = max_idle_time
        args['auto_connect'] = auto_connect

        DB.__init__(self, **args)

本文由美高梅游戏平台网站发布于鲜果干果,转载请注明出处:python 从数据库表生成model,pythonmodel

关键词:

上一篇:python学习之多线程,python多线程

下一篇:没有了

最火资讯