sqlalchemy是很强大的orm框架，一般的教程都是从新建一个表开始教，这种情况下我们能直接拿到表对象然后方便的操作，但是对于已经存在的表的操作，网上教程并不多，而实际上，对于测试同学来说，这种情况才是最普遍的。我在集成测试框架中数据库操作便使用了sqlalchemy，这里分享一下，顺便记录踩过的一些不那么明显却很致命的坑。

首先，我们可以创建一个自定义对象，它可以去组合sqlalchemy提供的种种功能。

from sqlalchemy.orm.exc import UnmappedClassError
from sqlalchemy.ext.declarative import declared_attr, declarative_base
from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import sessionmaker, class_mapper, Query

class DbRoot(object):

    def __init__(self, **kwargs):
        """
        orm基础db对象，通过实例化该对象得到db实例，然后创建类对象继承自db.Model，便可以对相应表进行操作
        :param kwargs: dialect 数据库类型
                        driver 数据库驱动
                        user 用户名
                        password 用户密码
                        host 数据库地址
                        port 端口
                        database 数据库名
        """
        url = '{dialect}+{driver}://{user}:{password}@{host}:{port}/{database}?charset=utf8'.format(**kwargs)
        engine = create_engine(url, echo=False)

        class Base(object):

            @declared_attr
            def __table__(cls):
                return Table(cls.__tablename__, MetaData(), autoload=True, autoload_with=engine)

        self._base = Base
        self.Model = self.make_declarative_base()
        self.session = sessionmaker(bind=engine)

    def make_declarative_base(self):
        """

        :return:
        """
        base = declarative_base(cls=self._base)
        base.query = _QueryProperty(self)
        base.query_class = Query
        return base

如代码所示，DbRoot对象接收数据库url的各项参数，然后创建引擎，组合declarative_base方法生成的Model对象(Model对象是Base对象的子类)，以及通过sessionmaker方法生成的session数据库会话对象(实例化它便可以产生一个数据库会话，相当于操作句柄)。
然后在make_declarative_base方法中，Model对象还组合了_QueryProperty，也就是query属性，用于查询操作，_QueryProperty是一个描述符对象，见下面代码。这样，我们在做比如query.filter_by等操作时，都是转给了get方法去执行，该方法最终调用了t也就是Model对象query_class方法(其实就是sqlalchemy.orm提供的Query方法)做查询，这里尤其注意，session参数传的是DbRoot对象的session属性，该属性一定要在这里实例化，这样每次操作才能重新生成一个session会话。

class _QueryProperty(object):

    def __init__(self, sa):
        self.sa = sa

    def __get__(self, obj, t):
        """
        这里一定要注意，session要每次重新生成，不然session会话会自动关闭，导致下一次操作句柄为空
        :param obj:
        :param t:
        :return:
        """
        try:
            mapper = class_mapper(t)
            if mapper:
                return t.query_class(mapper, session=self.sa.session())
        except UnmappedClassError:
            return None

现在，我们等于有了db.Model对象，然后我们要去映射表生成表对象。其实就是创建一个类继承自DbRoot.Model，然后将__tablename__属性设置为想要生成的表对象表名，这样便大功告成了。我这里采用了动态生成类的方式。

def gen_orm_class( db=None, table_name=None):
    """
    动态生成数据库表映射Model类
    :param db: db对象
    :param table_name: 表名称
    :return:
    """
    return type(
        table_name.title(),
        (db.Model,),
        {
            '__tablename__': table_name
        }
    )

有了这个基础，如果结合pytest使用的话会更方便，pytest有个很牛的功能叫fixture，我们可以把初始化数据库的操作做出一个scope为module的fixture

@pytest.fixture(scope='module')
def mysql(request, config_init):
    """
    mysql数据库操作实例
    :param request:
    :param config_init:
    :return:
    """
    db_roots = {}
    mysql_conf = config_init.get('mysql')
    databases = mysql_conf.pop('databases')
    dbs = request.module.config.get('mysql_dbs', {})

    for db, table in dbs.items():
        db_conf = databases.get(db)
        mysql_conf.update(db_conf)
        db_roots.update({table: gen_orm_class(db_name=db, db=DbRoot(**mysql_conf), table_name=table)})

    return db_roots

然后在测试用例中配置数据库连接信息。

# test_case.py
config = {
    'mysql_dbs': {
        'dev': 'user'
    }
}

这样在进入该模块(test_case.py)开始测试之前，pytest会根据配置自动初始化数据库连接生成{‘表名’: ‘DbRoot表对象’}结构的字典，之后在测试代码中，就可以通过mysql.get('user')拿到User表对象，从而对dev.User表进行查询操作。

    def test_user(self, mysql, user_id):
        user_tbl = mysql.get('user')
        # id是主键，所以可以通过get方法直接查询
        assert user_tbl.query.get(user_id)，'该用户{}不存在'.format(user_id)

有缘浏览到的人，希望可以帮助到你