sqlalchemy根据表名动态创建model类

版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/J_Object/article/details/80118657

作用如题，直接上代码吧，另外还支持 copy一张表的表结构，新建表并获得model对象

# coding: utf-8
import traceback
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
# 本地数据库
engineLocal = create_engine('mysql+pymysql://root:[email protected]:3306/lingyun?charset=utf8')
Base = declarative_base()
metadata = MetaData()
def dictToObj(results, to_class):
    """将字典list或者字典转化为指定类的对象list或指定类的对象
    python 支持动态给对象添加属性，所以字典中存在而该类不存在的会直接添加到对应对象
    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:
            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("传入对象非字典或者list")
    #     return None
def getModel(name, engine):
    """根据name创建并return一个新的model类
    name:数据库表名
    engine:create_engine返回的对象，指定要操作的数据库连接，from sqlalchemy import create_engine
    Base.metadata.reflect(engine)
    table = Base.metadata.tables[name]
    t = type(name,(object,),dict())
    mapper(t, table)
    Base.metadata.clear()
    return t
def createTableFromTable(name, tableNam, engine):
    """copy一个已有表的结构，并创建新的表
    metadata = MetaData(engine)
    Base.metadata.reflect(engine)
    # 获取原表对象
    table = Base.metadata.tables[tableNam]
    # 获取原表建表语句
    c = str(CreateTable(table))
    # 替换表名
    c = c.replace("CREATE TABLE " + tableNam, "CREATE TABLE if not exists " + name)
    db_conn = engine.connect()
    db_conn.execute(c)
    db_conn.close()
    Base.metadata.clear()
def getNewModel(name, tableNam, engine):
    """copy一个表的表结构并创建新的名为name的表并返回model类
    name:数据库表名
    tableNam:copy的表表名
    engine:create_engine返回的对象，指定要操作的数据库连接，from sqlalchemy import create_engine
    createTableFromTable(name, tableNam, engine)
    return getModel(name, engine)

本人测试人员，在做数据的验证的时候，反复调用pymy sql ，编写大量的 sql 语句，特别浪费时间，并且有时候业务比较复杂的时候，一句 sql 往往不能解决问题（当然如果上纲上线，所有业务逻辑都是可以用一句 sql 解决的，但那是建立在你的 sql 基础与时间成本上的）。所以我就产生了这个想法，是否可以借用ORM来实现数据的查询？当然我这个想法可能还不成熟，但不管如何，ORM还是我们测试人员必须会的一个知识点，所以就开始动手 1. SQLAlchemy 该框架偏向于底层，并且网上的资料也少，所以查阅了一些资料后就选择放弃 因业务需要从不同时期的 sql ite数据库中取数据，不同时期的数据库字段是不同的，但是以往用 sqlalchemy 管理数据库时的 model 是写死的，就会导致有些字段错误而报错，这时我们就要根据版本 动态 生成 model 众所周知python里的 type() 函数是可以查看一个实例的class 类 型，但其实 type() 也可以用来创建一个class 类 ，我们可以利用这一点 动态创建 model ，下面放出例子 # model .py from sqlalchemy import Column, String, Integer, 这个脚本将自动根据数据库中现有的表创建 SQLAlchemy 模型。请注意，此方法仅在数据库表结构已经存在时才有效。使用Python的 SQLAlchemy 可以根据 表名 自动创建模型。下面是如何做到这一点的示例。确保替换DATABASE_URI为您的数据库连接信息，并将。 http://blog.csdn.net/caoxing81/article/details/46639023 http://caochun.blog.51cto.com/4497308/1792304 http://stackoverflow.com/questions/17632980/ sql autocode-error-when-attempting-to-generate-schema 模型 ( model s) 是对数据抽象并提供通用访问接口的一种方式。 在大多数网络应用中，数据会被存储在一个关系数据库管理系统 (RDBMS) 中，也就是把数据格式化存储在由行与列组成的表格中，且能够跨表对数据进行比较。例如 My SQL 、Postgress、Oracle 和 MS SQL 。 为了基于数据库抽象出数据模型，我们需要使用一个叫做 SQLAlchemy 的 Python 包。 SQL Alche... 此错误表示在执行 SQL 语句时出现了问题，具体信息为：Not an executable object: 'select 1'。该错误是由 sqlalchemy 库中的 exc.ObjectNotExecutableError 引发的。 from flask_ sqlalchemy import SQLAlchemy HOSTNAME = '127.0.0.1' DATABASE = 'flask_ sqlalchemy ' PORT = 3306 USERNAME = 'root' PASSWORD = 'root' 参考：https://stackoverflow.com/questions/22355890/ sqlalchemy -multiple-foreign-keys-in-one-mapped-class-to-the-same-primary-key两张表layout和layout_union。layout_union表的layout_self_id和layout_other_id都关联到layou... 【Backend】 Sqlalchemy 统计表数目耗时少的方法 在使用flask- sqlalchemy 对多个表查询计数用len( 表名 .query.filter(Mtv.id).all())或者进行count操作时，出现了耗时严重、内存飙升的问题。 count = len( 表名 .query.filter( 表名 .id).all()) 解决：用 sqlalchemy 下的func就可以 from sqlalchemy import func count = db.session.query(func.cou """配置参数""" # sqlalchemy 的配置参数 连接到数据库 SQLALCHEMY _DATABASE_URI="[1 数据库 类 型]://[2 用户名]:[3 密码]@[4 数据库所在IP]:[5 端口号]/[6 数据库名]" #设置 sqlalchemy 自动跟踪数据库 SQLALCHEMY ...