深藏多层的意外提交耗时数月教你用AST和LLM规范数据库事务操作一个深藏多层的意外提交让我耗费了数月时间。本文将探讨为何数据库DB层必须掌控每一个提交和事务以及如何使用抽象语法树AST和代码检查工具linters来执行这些规则。2026年7月13日我为一个规范工作了好几个月最终所有相关方都批准了。我已将所有需要迁移的代码片段整理到了一个Notion数据库中。当处理第一个拉取请求PR时我看到了“db.commit()”这样的代码意识到这些“事务”在原子性的外衣下被提交了而我之前忘记考虑这些。这次问题并非出在对象关系映射ORM、参数化SQL语句或查询构建器上而是代码组织和抽象层面的严重失误导致事务内的代码失去了原子性。示例展示这些示例仅作说明之用代码已简化不包含任何框架或库的所有细微细节如会话创建等。隐藏的敌人class DBAccess:staticmethoddef create_records(records: List[DomainModel]):# 所有记录应作为一个事务处理with transaction():for r in records:DBAccess.create_main_records(r) # 调用辅助函数最终会调用 commit()# 另一个事务自动启动DBAccess.create_details_records(r)DBAccess.create_records(recs)这段代码在距离事务装饰器或上下文管理器两层以上的地方手动提交事务调用 create_main_records 时很难想到这个方法会进行提交操作。沉默的“假朋友”class DBAccess:staticmethoddef fetch_records(ids: List[int]) - List[DBModel]:db_models session.query(DBModel).filter(DBModel.id.in_(ids)).all()return cast(List[DBModel], db_models)with transaction():db_models DBAccess.fetch_records(ids)db_models[0].yo_mama_fat True # 这是一个无声的数据库写入操作# 上下文管理器退出时会帮你解决问题这段代码将数据库模型像普通领域模型一样传来传去设置属性看似只是赋值但实际上会触发数据库写入操作。一去不返class DBAccess:staticmethoddef fetch_records(ids: List[int]) - List[DBModel]:db_models session.query(DBModel).filter(DBModel.id.in_(ids)).all()return cast(List[DBModel], db_models)# 事务已被注释掉# with transaction():db_models DBAccess.fetch_records(ids)db_models[0].yo_mama_fat True# 请求结束数据消失得无影无踪# 就像程序员的爸爸出去买牛奶再也没回来这段代码既没有自动提交也没有事务处理会导致数据丢失。谁该负责不要责怪框架也不要责怪业务需求程序员是唯一的责任人。在很多情况下你可能不用为代码的现状负责但这次不行。任何做法都比随机提交要好。最后不管代码是不是你写的你都得负责解决这个烂摊子就像我这几个月一直在做的一样。我们能学到什么不要随意添加数据库会话或事务数据库抽象层应掌控事务和提交操作否则会像“撒盐哥”那样失败。不要在数据库层内外传递数据库模型。再次阅读第1点和第2点然后再读一遍把它们记在心里。不要在数据库层之外摆弄事务、提交、查询等操作连看都别看想都别想。不要手动提交尤其是在使用上下文管理器或装饰器的情况下你没那么厉害。不要把代码分散到辅助函数中原子性的多次写入操作应放在一个函数中代码重复能让原子性更明显也不会产生隐藏的问题。如何执行这些规则AST分析可以通过使用 [AST](https://docs.python.org/3/library/ast.html) 分析的自定义测试或者 [flake8](https://flake8.pycqa.org/en/latest/) 等代码检查工具来实现。无论选择哪种方式都应做到完全禁止手动调用提交操作。禁止在数据库访问层之外访问数据库会话。禁止在数据库访问层之外访问事务。禁止在数据库访问层之外导入数据库模型。使用AST进行禁止操作class TestDBBoundaries:def test_no_manual_commits():violations []for path, tree in parsed_source_files(): # 对源代码树进行 ast.parsefor node in ast.walk(tree):if not isinstance(node, ast.Call):continuefunc node.funcif ((isinstance(func, ast.Attribute) and func.attr commit) # session.commit()or (isinstance(func, ast.Name) and func.id commit)): # commit session.commit; commit()violations.append(f{path}:{node.lineno} manual commit())assert not violations, .join(violations)使用flake8进行禁止操作同样的遍历逻辑只是套上了flake8的“外衣”import astclass BanManualCommits:def __init__(self, tree):self.tree treedef run():for node in ast.walk(self.tree):if not isinstance(node, ast.Call):continuefunc node.funcif ((isinstance(func, ast.Attribute) and func.attr commit)or (isinstance(func, ast.Name) and func.id commit)):yield node.lineno, node.col_offset, DB001 manual commit(), type(self)注册方式如下[project.entry-points.flake8.extension]DB0 flake8_db_boundaries:BanManualCommits使用AST还是flake8在规则需要检查整个代码库或者不能修改共享的代码检查配置时使用AST测试。否则使用flake8或其他代码检查工具它们成为标准工具是有原因的。结合大语言模型LLM支持LLM可用于禁止数据库层返回任何数据库模型。AST和类型检查工具mypy都无法检测到这一点因为返回注释可能不准确cast() 函数会绕过类型检查且存在随机提交的代码库很难让人对类型系统有信心。因此这个检查可以在持续集成/持续部署CI/CD管道中通过LLM来完成。一个确定性的脚本会导出数据库访问层的公共函数/方法然后LLM回答一个简单的问题“是否有返回数据库模型而非领域模型的情况”结果如下是 - [人工审核](https://www.youtube.com/shorts/9arcEhyv_ZM)可能 - 人工审核否 - 通过领取工资总结数据库抽象层应掌控提交和事务其他做法都是因处理不当而采取的权宜之计。这样你就不用在发现一个随机提交后花费数月时间重构整个代码库来修复混乱的事务了。另外一定要读这本书[《领域驱动设计软件核心复杂性应对之道》](https://a.co/d/08hCw8SE)。想收听更多我的疯狂吐槽留下你的邮箱。订阅。如果你讨厌邮箱也可以使用 [RSS订阅源](/index.xml)。