因为多数python项目无需完整cqrs,真正需要的是意识到位与结构隔离:命令走明确入口、查询无副作用、状态变更显式触发。

为什么不用现成的 CQRS 框架?
因为多数 Python 项目压根不需要完整 CQRS——没有百万级写入、没拆分读写库、连命令和查询都还在同一个 models.py 里混着。硬套 django-cqrs 或 eventsourcing 反而让路由变模糊、测试变重、错误堆栈变长。
真正要的,是「意识到位 + 结构隔离」:命令走明确入口、查询不带副作用、状态变更不藏在 get_queryset() 里。
- 用
dataclass定义命令(不是Form,也不是Serializer),避免把校验逻辑塞进视图 - 查询一律通过专用函数或类方法,禁止在命令处理中顺手调用
filter()或select_related() - 所有状态变更必须显式触发,比如用
OrderPlaced而不是直接order.status = 'placed'
怎么组织命令处理器?
别写 CommandHandler 抽象基类,也别搞 handle() 反射调度。Python 的鸭子类型足够让你用最直白的方式隔离职责。
关键在命名和导入路径:把命令逻辑放在 commands/ 下,每个文件只做一件事,函数名就是动作本身。
立即学习“Python免费学习笔记(深入)”;
-
commands/place_order.py里只放place_order()函数,接收PlaceOrderCommand实例 - 函数内部不做 DB 查询,只调用
OrderRepository.create()和InventoryService.reserve()这类明确契约的依赖 - 如果某步失败(比如库存不足),抛出具体异常如
InsufficientStockError,而不是Exception或 HTTP 状态码 - 事务边界由调用方控制——通常在视图或 API endpoint 层用
@transaction.atomic包裹整个place_order()调用
查询层为什么不能复用 ORM QuerySet?
因为 Order.objects.filter(user=user).select_related('address') 看似简单,实则埋了三个坑:它隐含了 N+1 风险(后续加 prefetch_related 就失控)、耦合了模型字段(改个 related_name 就崩)、还偷偷执行了 SQL(在序列化前就查了库)。
正确做法是把查询逻辑收口为纯函数,输入 ID 或条件,输出 dict 或 NamedTuple。
- 写
queries/get_order_summary(order_id: int) -> dict,内部用原生 SQL 或values()+annotate(),确保只查必需字段 - 禁止在查询函数里调用
.save()、.delete()、.update()—— 一旦出现,立刻重命名该函数,加_mutating后缀提醒自己 - 缓存策略和查询分离:先保证
get_order_summary()正确,再考虑加@cache或 Redis key 拼接,别一上来就套cache_page
Django 中最容易漏掉的 CQRS 边界点
不是命令和查询的分离,而是信号(post_save)和事件的混淆。很多人以为发个 order_placed.send(sender=Order, instance=order) 就算完成事件驱动,其实只是把副作用从视图挪到了信号里,依然违反 CQRS 的“查询不可变”原则。
真正的边界在数据库写入之后、响应返回之前——这个窗口里,必须确保所有事件发布是幂等、可重试、不阻塞主流程。
- 用
transaction.on_commit()包裹事件发布,避免事务回滚时事件已发出去 - 事件数据必须序列化为字典,禁止传 model 实例(
instance.__dict__也不行,会带 _state 等内部字段) - 监听端(比如发邮件)必须独立部署或至少独立进程,不能和 Web 请求共用线程池,否则一个慢邮件拖垮整个订单接口
复杂点从来不在“怎么实现”,而在“谁负责清理失败事件”和“查询结果滞后几秒是否可接受”。这两个问题不提前对齐,代码写得再像教科书也没用。










