事件溯源的核心是保证事件写入与业务状态更新的原子性，需用本地事件表兜底、幂等apply、frozen dataclass定义事件、严格版本校验与顺序重放。

事件溯源不是加个 Event 类就完事

Python 服务里直接照搬 DDD 书里的事件溯源模式，大概率会在第 3 天遇到状态不一致、重放失败、调试困难三连击。核心问题不在“怎么建模事件”，而在于“谁来保证事件写入和业务状态更新的原子性”。SQLAlchemy 的事务边界默认不跨 session，而事件发布往往发生在 commit 后——这意味着数据库已提交，但事件可能卡在消息队列里丢弃了。

• 别在 after_commit 钩子里发事件：它不参与事务回滚，出错后状态和事件必然撕裂
• 用“本地事件表”兜底：把事件和业务数据一起写进同一张 events 表（带 processed 字段），再由后台任务轮询投递
• 避免在 __init__ 或 save() 中直接调用 publish_event()：这会让领域对象依赖外部消息系统，测试时根本绕不开网络

重放事件时 apply() 方法必须幂等且无副作用

重放不是“重新执行业务逻辑”，而是“用事件重建当前状态”。如果 apply() 里调了 requests.post() 或改了文件系统，重放一次就发三遍通知、删三次临时目录。

• apply() 只能操作内存中的聚合根属性，不能触发 I/O、不能修改其他聚合、不能调用 datetime.now()（要用事件自带的 occurred_at 时间戳）
• 聚合根构造函数里不要做任何非确定性操作；比如从配置中心读超时值、查缓存获取默认状态——这些值在重放时可能已变更
• 事件类字段必须全部是基本类型或可序列化的嵌套结构；别塞 lambda、threading.Lock 或数据库连接对象进去

用 dataclasses 定义事件比用 pydantic.BaseModel 更稳

pydantic 的验证和默认值逻辑在重放时会偷偷改数据：比如 created_at: datetime = Field(default_factory=datetime.utcnow)，重放旧事件时会覆盖原始时间戳。而 dataclasses 是纯容器，没隐藏行为。

Python Dependency Candidate Fetcher 2

根据包的能力以及 Jupyter 环境中的导入语句，从候选列表中挑选最适合用户表格工作流的 Python 包。

下载

• 事件类必须加 @dataclass(frozen=True)：防止重放中途被意外修改
• 序列化统一走 json.dumps(obj.__dict__)，别用 model.json()——后者可能注入额外字段或格式化时间
• 如果需要类型校验，用 typing.Annotated + typeguard 做运行时检查，而不是靠 pydantic 构造时“修正”输入

重放卡住？先看 event_id 和 version 是否连续

常见错误是重放失败后跳过某条事件，结果后续所有 version 校验都报 "expected 5, got 7"。事件溯源对顺序和完整性极度敏感，断点续传不是“从第 N 条继续”，而是“必须从上一个成功应用的 version + 1 开始”。

• 聚合根里维护 _version: int，每次 apply() 后自增；事件载荷里必须带 version 字段，两者严格比对
• 别用数据库自增 ID 当事件序号：不同聚合的事件混在一张表里，ID 不代表逻辑顺序
• 重放脚本要记录最后成功处理的 event_id（不是数据库主键），下次启动时从该 ID 的下一条查起

最麻烦的从来不是写事件，而是让重放过程在没有人工干预的前提下，既不丢也不重——这要求每条事件的语义边界足够干净，且所有时间、随机、外部依赖都被提前“冻结”进事件本身。

Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！