最近用 Flask 写了几个接口部署在服务器上，然后用 Pytest 来做测试，但遇到了问题，搞了大半天才把问题解决。
问题场景及原因
问题大概是这样的，我在本地环境用 Pytest 写代码来对服务器上 Flask 的接口进行测试，在测试删除接口的时候，第一步我在 Pytest 中会通过SQL插入数据到MySQL数据库，第二步再调删除接口完成删除用户。最后执行 Pytest 代码没跑通过，会返回用户不存在，但这个情况到数据库里查看，发现该用户实际是存在于MySQL中。
首先，这个接口是没啥大问题的，调注册接口或手工插入数据，再手工请求访问接口可以正常完成删除，只是我在 Pytest 代码中通过SQL插入数据会出现这个问题。
其次，怀疑自己 Pytest 代码写得有问题，但检查后发现没问题，并且加了查询操作发现能查回这个新插入的数据。
最后，在各种排查后，在 Flask 接口服务中，也加了查询操作，但发现压根没查回这个新插入的数据。
于是，感觉接口环境下通过 Python 连接 pymysql ，查询数据库时，获取到的似乎不是最新数据，所以调用户删除接口才会提示用户不存在。接着，在网上查了下资料，发现是事务的隔离级别导致的这个问题。
MySQL默认事务隔离级别是
REPEATABLE READ，当我在本地 Pytest 代码中利用SQL插入用户数据并提交 commit 后，接口环境下的事务A不会读取到我本地环境下的事务B更新的信息，即便事务B已提交，而事务A每次查询到的数据都是最开始创建事务查询到结果的快照，事务A一直没进行更新。更详细的解释说明，可参考文末列出的网上资料。
问题代码
import pymysql
class MysqlDb():
def __init__(self, host, port, user, passwd, db):
# 建立数据库连接
self.conn = pymysql.connect(
host=host,
port=port,
user=user,
passwd=passwd,
db=db
)
# 通过 cursor() 创建游标对象，并让查询结果以字典格式输出
self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
def __del__(self): # 对象资源被释放时触发，在对象即将被删除时的最后操作
# 关闭游标
self.cur.close()
# 关闭数据库连接
self.conn.close()
def select_db(self, sql):
"""查询"""
# 检查连接是否断开，如果断开就进行重连
self.conn.ping(reconnect=True)
# 使用 execute() 执行sql
self.cur.execute(sql)
# 使用 fetchall() 获取查询结果
data = self.cur.fetchall()
return data
def execute_db(self, sql):
"""更新/新增/删除"""
try:
# 检查连接是否断开，如果断开就进行重连
self.conn.ping(reconnect=True)
# 使用 execute() 执行sql
self.cur.execute(sql)
# 提交事务
self.conn.commit()
except Exception as e:
print("操作出现错误：{}".format(e))
# 回滚所有更改
self.conn.rollback()
解决办法
根据以上说明，要解决当前这个问题，我们可以通过以下方式来处理。
方法1：修改MySQL的事务隔离级别
方法2：每次查询操作后，都进行commit()提交事务。
方法3：Python创建pymysql连接时，设置autocommit=True，即让其操作后自动提交事务。
在这里，我们最好在方法2和方法3任选一个方法来处理就行了，不建议去修改MySQL的事务隔离级别。
- 每次查询后进行提交事务
def select_db(self, sql):
"""查询"""
# 检查连接是否断开，如果断开就进行重连
self.conn.ping(reconnect=True)
# 使用 execute() 执行sql
self.cur.execute(sql)
# 使用 fetchall() 获取查询结果
data = self.cur.fetchall()
# 提交事务
self.conn.commit()
return data
- 设置 autocommit=True 自动提交事务
def __init__(self, host, port, user, passwd, db):
# 建立数据库连接
self.conn = pymysql.connect(
host=host,
port=port,
user=user,
passwd=passwd,
db=db,
autocommit=True
)
# 通过 cursor() 创建游标对象，并让查询结果以字典格式输出
self.cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
使用以上方法后，再次测试，发现问题已顺利得到解决。
参考资料：
为什么pymysql重连后才能查到被其他地方修改的数据 pymysql缓存？
记一次pymysql查询不到表中最新插入的数据的问题
网友评论