如何在Python自动化中优雅处理数据库死锁测试-Python教程-PHP中文网

死锁可复现：用两个线程分别按相反顺序更新同一组记录（如A: ID=1→2，B: ID=2→1），显式开启事务且不提交，配合MySQL的innodb_print_all_deadlocks=ON和捕获错误码1213即可稳定触发并验证重试逻辑。

如何在python自动化中优雅处理数据库死锁测试

怎么复现死锁来验证重试逻辑是否生效

不先让死锁稳定出现，后续所有重试、日志、监控都无从验证。关键不是“避免”，而是“可控地触发”。

用 threading.Thread 启两个并发线程，各自执行一个事务：线程 A 先 UPDATE ID=1 再 UPDATE ID=2；线程 B 反过来先 ID=2 再 ID=1
确保两事务都显式开启（如 BEGIN 或 ORM 的 session.begin()），且中间无 COMMIT
在 MySQL 中开启 innodb_print_all_deadlocks = ON，死锁发生后立刻能在错误日志里看到完整堆栈和持锁 SQL
Python 捕获异常时检查 exc.args[0] == 1213，这是唯一可靠的死锁判定依据，别依赖字符串匹配

为什么不能只靠改 `isolation_level` 配置来防死锁

把 SQLALCHEMY_ENGINE_OPTIONS['isolation_level'] 设成 READ COMMITTED 或 SERIALIZABLE，对解决 1213 错误基本无效。

isolation_level 只影响新连接的默认级别，不会覆盖已开启事务的行为
MySQL 的 1213 死锁根源是加锁顺序不一致，不是隔离级别低——REPEATABLE READ 下间隙锁多，但即使设成 READ COMMITTED，只要 A 锁 1→2、B 锁 2→1，照样死锁
盲目设 SERIALIZABLE 反而扩大锁范围，提高冲突概率；SQLite 则直接忽略该配置
真正起作用的是应用层控制：统一按主键升序查、锁、更新，例如 query.filter(id.in_([101,102])).order_by(Model.id).with_for_update()

测试中如何避免线程卡死导致 pytest 挂住

pytest 不干预线程内部逻辑，一旦测试函数里出现无限等待，整个 worker 就 hang 住，CI 流水线会超时失败。

python-flask-vue-creator

创建 Flask + Vue.js 全栈项目，采用单端口架构、uv 依赖管理、结构化日志和生产级前端设计。当用户要求以下操作时触发：创建新的 Flask+Vue 项目、搭建 Web 应用、构建全栈 Python+Vue 应用、初始化 Flask Web 项目、设置单端口 Flask 前后端项目。触发词：新建项目、创建项目、搭建项目、Flask项目、Vue项目。

下载

所有 thread.start() 后必须跟 thread.join(timeout=2)，timeout 不能省略或为 None
避免裸用 threading.Lock，改用带超时的 lock.acquire(timeout=0.5)，并检查返回值是否为 True
不要在测试里做 HTTP 调用、文件 IO 或 sleep(>0.1)，这些会拉长事务时间，放大死锁概率
若需验证锁行为，用 with_for_update(nowait=True) 主动抛 OperationalError（MySQL 错误码 1205），比等死锁更易断言和清理

重试机制必须带退避且校验业务状态

捕获到 1213 后直接重试一次，大概率再次撞上同样死锁路径——因为竞争条件没变。

必须用指数退避：第一次等 0.1s，第二次 0.2s，第三次 0.4s，最多 3 次
重试前要重新查询关键字段（如库存余额、订单状态），确认上次操作是否其实已成功提交（网络延迟或事务回滚不透明）
不要在 ORM session 里重试：session 对象在死锁后已处于不可用状态，必须新建 session 或显式 session.rollback()
重试逻辑应封装进装饰器或上下文管理器，避免每个业务函数重复写 try/except 套路

死锁测试最易被忽略的点：事务边界是否真由测试代码控制。ORM 自动 commit、连接池复用、异步 event loop 中的 session 生命周期，都会让“你以为的事务”实际跨了多个物理连接或语句块。验证时务必打日志输出每条 SQL 和其所属 connection ID。