接口测试时，测试数据有多种方式来存放（比如yaml、excel、MySQL、甚至txt文件等等），使用哪种方式需要根据接口的实际情况（请求数据量大小）来决定。当然也可以写死在代码里，不过从可维护性的角度来看这显然是一个糟糕的做法。还有另外一个场景，大多数情况下，POST或PUT接口会向业务数据库（MySOL）中插入数据，如果想进一步增强接口校验的可信度，除了对返回的response body做校验之外，业务数据库中是否完成插入数据的动作也可以作为一个校验点。

无论是用MySOL来存放测试数据还是校验业务数据库（MySOL）中是否插入数据，都需要做到使用Python脚本完成MySQL的增删改查操作。MySQL操作有两种方式，第一种最直接的方式是用SQL语句拼接成字符串，然后执行SQL，这种方式同样面临维护性的问题（如果表名发生了变化，需要在引用表名的所有位置全部完成替换），同时大量SQL字符串的存在会使得代码看起来异常臃肿。第二种是使用ORM框架，ORM框架把MySOL表和Python的类一一对应，MySQL的增删改查操作对应类中的各个方法，这大大降低了维护难度。

Python中，SQLAlchemy是最有名的ORM框架之一。

安装

pip3 install sqlalchemy

SQLAlchemy架构：

Schema / Types       　　　　　架构和类型SQL Expression Language      封装好的 SQL 语句Engine 　　　　　　　　　　　　 框架引擎Connection Pooling 　　　　　数据库连接池Dialect 　　　　　　　　　　　根据用户的配置，调用不同的数据库 API(Oracle, postgresql, Mysql) 并执行对应的 SQL语句

SQLAlchemy本身无法操作数据库，其必须依赖pymsql等第三方插件，从而实现对数据库的操作，如：

pymysql    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

如果和我一样用的是Python3，需要先确保pymysql已经安装。

SQLAlchemy操作数据表CURD

直接上代码

创建数据表（model.py）：

from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmaker# 创建对象的基类Base = declarative_base()# 初始化数据库连接engine = create_engine('mysql+pymysql://root:root@localhost/0501_db',                       echo=True,  # 程序运行时反馈执行过程中的关键对象，包括ORM构建的sql语句                       max_overflow=0,  # 超过连接池大小外最多创建的连接                       pool_size=5,  # 连接池大小                       pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错                       pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）                       )'''创建数据表的映射类'''# 定义Users对象class Users(Base):    # 表的名字    __tablename__ = 'users'    # 表的结构    id = Column(Integer, primary_key=True)    age = Column(String(3))    name = Column(String(20))# 定义Cat对象class Cat(Base):    __tablename__ = 'cat'    id = Column(Integer, primary_key=True)    name = Column(String(20))    color = Column(String(20))# 创建表到数据库表中Base.metadata.create_all(engine)

declarative_base()是sqlalchemy内部封装的一个方法，Base作为所有映射类的父类，将在子类中把声明式映射过程作用于其子类。

从上面的代码不难总结出：

一个类对应数据库中的一张表，类的数据属性对应了表中的字段名，这个类称为映射类。根据映射类创建出一个个的对象，每个对象对应了表中的一条实际的数据。

图形化语言：

执行代码，终端查看数据表：

有了数据表，怎么去操作呢？根据上面所说“根据映射类创建出一个个的对象，每个对象对应了表中的一条实际的数据”，创建一个对象就可以了吗？比如这样：

user_1 = Users(age='18', name='yangge')

当然是不可以的，此时，user_1对象只是在此刻环境的内存中有效，并没有在表中真正生成数据。

要想生成数据到表中，需要创建一个和数据库沟通的会话对象，利用这个会话对象对数据库中的表进行CURD操作。

数据表的CURD操作：

sqlalchemy使用session用于创建程序和数据库之间的会话，所有对象的载入和保存都需要通过session对象。

在model.py中追加如下代码：

from sqlalchemy.orm import sessionmaker# 把当前的引擎绑定给这个会话DBSession = sessionmaker(bind=engine)# 添加# 创建Session对象session = DBSession()# 创建User对象user_1 = Users(age='18', name='zhangwu')user_2 = Users(age='19', name='zhangliu')user_3 = Users(age='20', name='zhangjiu')# 添加到session# session.add(user_1)session.add_all([user_1, user_2, user_3])# 提交session.commit()# 关闭sessionsession.close()

执行代码，终端查看数据表：

查询：

# 创建sessionsession = DBSession()# 利用session创建查询，query(对象类).filter(条件).one()/all()# one()返回一个记录，all()返回所有符合条件的记录列表user = session.query(Users).filter(Users.id == '2').one()user_list = session.query(Users).all()print('type:{0}'.format(type(user)))print('name:{0}'.format(user.name))print('type:{0}'.format(type(user_list)))print('name:{0}'.format(user_list[2].name))# 关闭sessionsession.close()

控制台输出：

更新&删除：

# 更新session = DBSession()user_result = session.query(Users).filter_by(id='2').first()user_result.name = "lishi"session.commit()session.close()# 删除session = DBSession()user_del = session.query(Users).filter_by(id='1').first()session.delete(user_del)session.commit()session.close() 

由于关系型数据库的多个表还可以用外键实现一对多、多对多等关联，相应地，作为ORM框架的SQLAlchemy也支持多表的CURD。

源码：