我已经在一个规范上工作了几个月。最终……每个利益相关者都批准了。我在一个 Notion 数据库中 catalog 了需要迁移的每一段代码。我正在处理第一个 PR，然后我看到了…… db.commit() '……他妈的……'该死的“事务”在穿着一件原子性外衣的同时被提交了。“靠，靠，靠，我怎么会错过这个。”翻看我的笔记……我看到……“调查随机提交”我忘了考虑它们……ORMs 可能会炸掉你的腿，这不是我想说的。参数化的 SQL 语句可以是你的救星，这不是我想说的。查询构建器可以是一个不错的折中方案，这也不是我想说的。这段 rant 是关于如何搞砸代码组织和抽象，使得包裹在事务中的代码不是原子的。示例展品 这些旨在说明，因此代码已被清理，不包含任何框架/库的所有小细节。会话正在创建等。 隐藏的敌人类 DBAccess: @staticmethod def 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) 这段代码在与事务装饰器/上下文管理器相距 2 个以上级别的位置搞乱手动提交。因此，当从管理器调用 create_main_records 时，没有人知道该方法会提交。沉默的朋友类 DBAccess: @staticmethod def 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 # 这是一个静默的 DB 写入 # 上下文管理器 exit 保存了你的命 Feng- 代码在传递 DB 模型，把它们当作普通域模型来对待，设置属性似乎只是在设置值，但在所有这些背后，触发了 DB 写入。父亲的牛奶之旅类 DBAccess: @staticmethod def 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 # 请求结束，数据消失在虚空中 # 就像程序员的父亲去买牛奶，然后不再回来 这段代码没有自动提交，也没有事务，导致了数据丢失。想听听我更多的无理 rant 吗？留下你的邮箱。或者如果你讨厌电子邮件，可以使用 RSS 源。 谁负责？ 不要责怪框架 不要责怪业务需求。 程序员是唯一一个负有责任的人。在很多情况下，你可以被免除代码状态的责任，但在这种情况下却不行。 随机提交总比随意提交要强。最后，不论你是否编写了这段代码（我的情况），你的屁股都在修复这些问题时处于危险之中。就像我这几个月一样。我们在学习什么？ 不要像撒盐一样点缀 DB 会话或事务。如果你这样做，期望像他一样轰然倒塌。数据库抽象层拥有事务和提交。 不要在 DB 层内外传递 DB 模型。重新读一遍 1+2，然后再读一遍。吸收它们。 不要在 DB 层外搞乱事务、提交、查询或其他任何东西。甚至不要看它，甚至不要考虑它。 不要手动提交。 尤其是如果你使用上下文管理器或装饰器。你可不是那种人。 不要在帮助程序中拆分代码。原子多写 = 一个函数。再写 3 行该死的插入。重复性使原子性可见，你不会创造隐藏的敌人。否则，要小心。 你如何强制执行这些教训？ AST 分析：通过使用 AST 分析的自定义测试或通过 flake8 /linters。选择哪种方式都应该： 完全禁止手动调用提交 禁止在 DB 访问层外部访问 db 会话 禁止在 DB 访问层外部访问事务 禁止从 DB 访问层之外导入 DB 模型 禁止使用 AST 类 TestDBBoundaries: def test_no_manual_commits(self): violations = [] for path, tree in parsed_source_files(): # ast.parse 遍历你的源树 for node in ast.walk(tree): if not isinstance(node, ast.Call): continue func = node.func if ( (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} 手动提交()') assert not violations, ' '.join(violations) 禁止使用 flake8 同样的遍历，穿着 flake8 的外衣