sqlalchemy 利用已存在表結(jié)構(gòu)建立新表并生成新的model類的方法：

# coding: utf-8
import traceback

import datetime
from sqlalchemy import (BigInteger, Column, DateTime, Integer, MetaData, String, Table, create_engine, text)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import mapper
from sqlalchemy.schema import CreateTable


# 本地?cái)?shù)據(jù)庫(kù)
engineLocal = create_engine('mysql+cymysql://root:password@192.168.0.111:3306/pop')


BaseSQL = declarative_base()
metadata = MetaData()


def dictToObj(results, to_class):
    """將字典list或者字典轉(zhuǎn)化為指定類的對(duì)象list或指定類的對(duì)象
    python 支持動(dòng)態(tài)給對(duì)象添加屬性斟或，所以字典中存在而該類不存在的會(huì)直接添加到對(duì)應(yīng)對(duì)象
    """
    if isinstance(results, list):
        objL = []
        for result in results:
            obj = to_class()
            for r in result.keys():
                obj.__setattr__(r, result[r])
            objL.append(obj)
        return objL
    else:
        try:
            obj = to_class()
            for r in results.keys():
                obj.__setattr__(r, results[r])
            return obj
        except Exception as e:
            print(e)
            traceback.print_exc()
            return None
    # else:
    #     print("傳入對(duì)象非字典或者list")
    #     return None


def getModel(name, engine):
    """根據(jù)name創(chuàng)建并return一個(gè)新的model類
    name:數(shù)據(jù)庫(kù)表名
    engine:create_engine返回的對(duì)象读串，指定要操作的數(shù)據(jù)庫(kù)連接个束，from sqlalchemy import create_engine
    """
    BaseSQL.metadata.reflect(engine)
    table = BaseSQL.metadata.tables[name]
    t = type(name, (object,), dict())
    mapper(t, table)
    BaseSQL.metadata.clear()
    return t


def createTableFromTable(name, tableNam, engine):
    """copy一個(gè)已有表的結(jié)構(gòu)，并創(chuàng)建新的表
    """
    metadata = MetaData(engine)
    BaseSQL.metadata.reflect(engine)
    # 獲取原表對(duì)象
    table = BaseSQL.metadata.tables[tableNam]
    # 獲取原表建表語(yǔ)句
    c = str(CreateTable(table))
    print(c)
    print(tableNam, name)
    # 替換表名
    c = c.replace('"', '').replace('CREATE TABLE {}'.format(tableNam), "CREATE TABLE if not exists {}".format(name))
    print(c)
    db_conn = engine.connect()
    db_conn.execute(c)
    db_conn.close()
    BaseSQL.metadata.clear()


def getNewModel(name, tableNam, engine):
    """copy一個(gè)表的表結(jié)構(gòu)并創(chuàng)建新的名為name的表并返回model類
    name:數(shù)據(jù)庫(kù)表名
    tableNam:copy的表表名
    engine:create_engine返回的對(duì)象桨踪，指定要操作的數(shù)據(jù)庫(kù)連接，from sqlalchemy import create_engine
    """
    createTableFromTable(name, tableNam, engine)
    return getModel(name, engine)


def drop_table(name, engine):
    drop_sql = "DROP TABLE if exists {}".format(name)
    db_conn = engine.connect()
    db_conn.execute(drop_sql)
    db_conn.close()


if __name__ == "__main__":
    t = datetime.date.today().__str__()
    name = "test_{}".format(''.join(t.split('-')))
    # tableNam = 'interaction_CM191001'
    # getNewModel(name, tableNam, engineLocal)
    createTableFromTable(name, 'security_daily_evaluation', engineLocal)