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):

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)