写这篇之前我先说说背景。在Python社区聊领域驱动设计DDD大多数人第一反应是这不Java那套吗我们Python不搞这些另一部分人看过一两本讲DDD的书结果满脑子都是Entity、Value Object、Aggregate、Domain Event这些名词落到代码里完全不知道怎么组织目录和依赖。我这些年用Python做过交易系统、订单中台、也做过几个内部工具平台全程把DDD当第一性原理来使踩了不少坑也彻底尝到甜头。这篇不是翻译DDD蓝皮书的笔记而是一篇从理论到生产级实践的完整复盘——哪些概念必须落地、哪些可以扔到一边、Python的特性在这个体系里能占多少便宜以及从烂代码渐进重构的实测路径。如果你正准备在Python项目里引入DDD或者已经在写但总觉得别扭这篇应该能帮你省下几个月试错时间。1. 先泼冷水DDD不是包治百病的银弹Python也不是圈子里有一个普遍误区觉得用了DDD就高级了或者觉得自己项目小不配用DDD。这俩都是对的也都不对。我见过有人在只有两张表的小工具里强行搞出六个聚合根和一套事件总线结果代码量翻了三倍改任何需求都要动十个文件也见过核心业务逻辑复杂到几百行的一个大Service天天出线上事故却从来没人动过重构的念头——因为没人带过他们用领域思维拆解过问题。DDD真正的适用场景是业务规则复杂、存在多条业务线的串联、需求会长期演化、以及多个模块之间互相牵扯状态。如果你的项目只是简单的CRUD事务脚本反而简单清晰强行DDD属于自找麻烦。反过来当你发现Service层变成了一个垃圾场——一堆服务互相调用、实体几乎只是数据容器、每加一个需求就要改五六个Service方法的时候就是DDD进场的最佳时机。Python不平白送太多东西但它有个非常宝贵的特性对类型系统的宽容和对表达力的释放让你能在不引入Java那样沉重框架的前提下用轻量代码实现DDD的核心骨架。再加上dataclasses、attrs、Protocol、泛型注解这些工具Python完全可以承载聚合、值对象、领域事件、仓储这些模式而且写得比Java更顺手——关键是你得知道哪些该用哪些不该用。我在下文会反复提到一个观点DDD落地的核心不是用了什么工具和框架而是保证你的核心代码不依赖基础设施、不被框架带偏让领域逻辑成为整个系统里最稳定、最容易被测试、最能表达业务含义的那一层。Python在这点上优势很明显不需要依赖注入容器不需要XML配置不需要AOP框架你完全可以用最朴素的方式组织出清晰架构。2. 从理论到战术建模先别写代码先画出业务边界2.1 限界上下文不是一张架构图而是一道行政命令很多团队做DDD第一步先画一堆框图把每个子域画得漂漂亮亮然后就没有然后了。问题在于限界上下文Bounded Context本质上是团队协作和代码归属的边界它直接决定了你在哪个模块里可以自由更改模型。落不到代码所有权上的限界上下文就是PPT。我在一个订单系统里实践过。当时我们把订单、库存、支付、营销四个子域拆成了四个限界上下文划分标准不是数据库表因为是同一个库也不是微服务当时是单体而是每个领域语言的封闭性。订单里出现商品快照这个概念库存里出现库存实物这个概念颜色、含义完全不同任何一方都不允许直接拿对方的模型类来用。哪怕数据库里都是product_id字段通过代码的模块归属就把它们隔离了。实操建议是你至少要在Python项目的目录结构上体现出这种边界。在模块边界未落实之前任何DDD都只是改改类名的编码风格不是架构演进。你可以用src/order、src/inventory、src/payment这样的顶层模块来切割即便它们仍然在同一条进程里模块间的调用也要经过明确的接口而不是import一个数据类全当传参。2.2 战术建模实体、值对象、聚合到底各自管什么战术建模这部分概念容易懂难的是判断一个对象到底该建模成Entity还是Value Object。我的判断标准特别简单它有没有身份这个身份是否需要被持续追踪有就做Entity没有就做Value Object。订单是Entity因为它有订单号会被状态流转引用并追踪订单地址是不折不扣的Value Object——合同地址从A改到B你绝对不是说原来的地址对象发生了变化而是整体替换成一个新的地址快照。Python里用dataclass的frozenTrue就能很优雅地表达值对象还能顺带解决哈希和相等性比较的问题。聚合Aggregate则是重头戏。聚合不是把所有关联对象捆在一起而是为了一致性划出一条边界而且边界以内的事务提交是原子的。我在实际项目中见过最典型的问题把订单和订单项做成两个聚合订单项增删直接用ORM的方式操作结果在并发下经常出现订单总金额与明细不一致。正确做法是把Order作为聚合根订单项只是聚合内部的子实体对订单项的任何修改都必须经由Order这个入口。下面是简化代码示例用Python的dataclass实现dataclass class OrderLine: product_id: str quantity: int unit_price: Money dataclass class Order: id: OrderId customer_id: CustomerId lines: list[OrderLine] status: OrderStatus def add_line(self, product_id: str, quantity: int, unit_price: Money) - None: if not self.can_change(): raise OrderImmutableError(订单已支付不能修改商品) self.lines.append(OrderLine(product_id, quantity, unit_price)) self.recalculate_total() def recalculate_total(self) - None: self.total Money(sum(line.unit_price.amount * line.quantity for line in self.lines))这段代码里add_line是聚合根的唯一入口任何外部代码都不允许直接往lines列表里塞数据。所有不变量例如只有未支付订单才能改明细都收拢在这个函数内。这才是聚合的意义——不是把类包得漂亮而是把状态变更的规则锁在根里。2.3 领域服务比起本服务才是你应该多写的很多Python开发者听DDD的领域层立刻想到的是domain/model.py里一堆装饰了dataclass的类而真正的业务逻辑还是写了一堆上帝Service。这是DDD最容易被理解偏的地方你可以在领域层写领域服务这类服务的定位是多个聚合或多个值对象之间的协作它属于领域语言的一部分而不是被基础设施牵着鼻子走的应用服务。比如计算订单是否免运费这种逻辑它既不属于订单聚合也不属于客户聚合——放在domain/service.py里面是一个很纯粹的领域服务。而创建订单并调用支付接口这种编排多个聚合动作的职责应该放在应用服务层但它要做的是编排不是塞业务计算。两层的核心区别是领域服务只依赖领域模型和领域事件不依赖数据库、不依赖外部HTTP、不依赖task queue。下层的调味代码既要能表达业务规则又要不掺入任何基础设施的杂质这是DDD的另一个名字让核心业务代码保持最干净的纯度。3. 搭建第一个聚合根依赖方向比代码量重要一百倍3.1 从数据库倒推建模还是从业务倒推建模我见过一份典型的伪DDD代码模型类长得非常像数据库表字段Repository直接暴露model.save()应用服务里到处是session.commit()。这种写法根本原因是——很多教程教你先建表再反向生成类再套上DDD的皮。真正的DDD是反过来的先从业务规则出发建模然后让持久化层去适配模型。这句话翻译成技术动作就是模型层不能依赖ORM尤其不能依赖ORM的基类和查询语法。在Python里最痛苦的就是SQLAlchemy这类ORM的诱惑——声明式基类一继承声明关系一用你的模型马上就变成一张表的设计产物。如果你做的事足够复杂这种模型会在未来某一天让你崩溃。生产级做法是把模型定义和表映射彻底分开。模型用纯dataclass定义SQLAlchemy的表映射单独放在infrastructure/persistent.py里用mapper_registry或者走独立的dataclass映射方式。是的这样做会有一定代码量冗余但它换来的是领域模型不被任何数据库方言污染将来哪怕你从PostgreSQL换到DynamoDB核心业务代码动都不用动。这种依赖倒置的收益在业务复杂度起来之后会特别明显。3.2 值对象与实体的Python式写法我偏好的方式是值对象用frozen dataclass实体用普通dataclass再加一个内部_id。这里要强调一个区别Python的dataclass默认生成的__eq__是逐字段比对对实体来说这是错误的语义。实体相等性应该基于身份也就是ID一致。所以我通常会dataclass class Order: id: OrderId # ... fields def __eq__(self, other): if not isinstance(other, Order): return NotImplemented return self.id other.id def __hash__(self): return hash(self.id)而值对象比如Money我会这样定义dataclass(frozenTrue) class Money: amount: Decimal currency: str def __add__(self, other: Money) - Money: if self.currency ! other.currency: raise CurrencyMismatchError(...) return Money(self.amount other.amount, self.currency)这看着简单但解决了大量隐性问题相等性、哈希化、不可变性、运算校验。Money混用decimal和float是生产事故高发地建议全文统一用Decimal模型边界里禁止float当钱用。这也是个没法讨价还价的规范。3.3 聚合根的完整性校验放到哪不变量校验在DDD里有明确的归属聚合根对自己内部的一致性负责。外部条件是否满足比如库存是否足够、支付渠道是否可用属于应用服务或领域服务的编排职责不该堆进聚合根的setter里。我写订单的常见完整性问题有人把库存扣减和订单创建放在同一个事务里同时把库存是否足够的判断塞进Order.add_line。这就把聚合边界捅破了——订单聚合不该知道库存那是库存上下文的事。正确的做法是应用层先调用InventoryService.ensure_available再把确认结果传给Order.create让订单自己决定是否接受这个结果。虽然在同一个事务里看起来多绕一圈但这恰恰保证了每个聚合边界都只做自己管辖内的事。4. Repository与基础设施解耦你的ORM不该出现在领域层4.1 Repository接口到底要不要定义在domain层Repository是DDD里被误解最多的模式之一。很多人以为它只是一个包装了数据库查询的工具类放在service下面随便用。我在生产级的Python项目里坚持把Repository接口定义在domain层而不是implementation层。为什么因为domain层需要表达我有哪些数据资源可以用但它不关心你从MySQL拿还是从Redis缓存拿。比如:class OrderRepository(Protocol): def get(self, order_id: OrderId) - Order: ... def save(self, order: Order) - None: ...这个Protocol只定义有什么方法、参数类型是什么、返回什么”不定义如何持久化。然后infrastructure层再去写SQLAlchemy实现。class SqlAlchemyOrderRepository: def __init__(self, session_factory): self._session_factory session_factory def get(self, order_id: OrderId) - Order: with self._session_factory() as session: record session.execute( select(OrderTable).where(OrderTable.id str(order_id)) ).scalar_one() return self._to_domain(record) def save(self, order: Order) - None: with self._session_factory() as session: record self._from_domain(order) session.merge(record)你会发现这里出现了两个模型持久化模型record和领域模型Order。它们的长相可能差不多但分属两个层——这正是依赖倒置的体现高层模块不依赖低层模块两者都依赖抽象和适配器。可能有人觉得这太啰嗦了但现实情况下当你的查询需求开始分裂一个来自OLTP写路径一个来自OLAP读模型一个来自报表聚合的时候单一ORM映射模型的维护成本会指数增长。提前分流收益远大于那点样板代码。4.2 事务边界UnitOfWork还是直接在Repository里处理Python中事务边界常常被忽略。默认的Session也能搞定但我更推荐轻量级UnitOfWork模式原因只有一个当多个Repository需要协同修改时事务边界必须清晰不能依赖于调用方记得调用的顺序。简单实现可以用contextlib.contextmanager包装class UnitOfWork: def __init__(self, session_factory): self._session_factory session_factory def __enter__(self): self.session self._session_factory() return self def __exit__(self, exc_type, exc_val, tb): if exc_type: self.session.rollback() else: self.session.commit() self.session.close() property def orders(self) - OrderRepository: return SqlAlchemyOrderRepository(self.session)然后在应用服务里明确写出事务的开始和结束def create_order_application_service(command: CreateOrderCommand) - None: with uow.begin() as uow: customer uow.customers.get(command.customer_id) order customer.place_order(command.items) uow.orders.save(order)这个写法让应用服务不用知道commit和rollback藏在哪所有仓储绑定在同一个会话上事务边界一目了然。这也是生产级DDD里最容易被忽视的最后一公里——模型画得很纯但事务边界乱了那一切白搭。4.3 仓储与查询的分离Command查询、Query查询这里我给一个额外建议把一个仓储对象拆成两边。写操作类仓储保持领域语义读操作查询直接面向页面或报表模型。Python可以很自然地用类方法或构造器参数区分。很多人把一个Repository揉成一个大杂烩既有get、save又有search_like、top_n、page_query最后导致整个仓储接口膨胀domain层被各种查询结果DTO绑架。我一般是Command仓储领域层依赖只暴露get/save/delete等方法。Query服务/读模型直接面向界面的数据库查询返回扁平dict或专用DTO不经过ORM的delivery层。这一条做好了你的应用层既不会出现对ORM的直接依赖也不会出现一个仓储返回十几个不同形态数据的怪事。5. 事件驱动与消息机制从领域事件到异步解耦的实践5.1 领域事件是纯Python对象不是消息队列领域事件在DDD里的正确用法是**聚合在状态变更时记录一组已经发生的事实应用层把事件发布出去其他限界上下文通过订阅做出反应。**它和消息队列不是一回事。消息队列是基础设施领域事件是先于基础设施存在的领域概念。我用Python实现一个简单的领域事件集合通常是聚合根内部维护一个events列表dataclass class OrderPlaced: order_id: str occurred_at: datetime class Order: def __init__(self): self.events: list[DomainEvent] [] def place(self): # 业务校验... self.status OrderStatus.PLACED self.events.append(OrderPlaced(order_idstr(self.id), occurred_atdatetime.utcnow()))重点在于order.place()完成后事件已经存在于聚合内部应用层决定何时把这些事件调度出去但事件的发生和存储是领域层的职责。这样可以保证事件不丢、不重放。5.2 事件发布的最佳时机事务提交后初学DDD最容易踩的坑在聚合方法内部立刻调用事件发布或者在一个还没提交的事务里就把事件发给外部系统。这会导致两个严重问题一是事件发出去了但事务回滚了二是消息队列里的消费者处理时发现数据库里没有对应记录。正确的顺序是应用服务开启UoW事务修改聚合根收集领域事件仓储保存聚合同一事务内事务提交成功后再遍历聚合收集的事件并发布到消息总线如果发布失败可以做本地事件表补偿的outbox模式我倾向于先做in-process事件总线起步等真正出了多服务需求时再换消息队列。而且事件总线的接口保持极度简单class EventBus(Protocol): def publish(self, event: DomainEvent) - None: ...这样是不是消息队列、是不是异步都会变得很可替代。5.3 异步事件处理与事务穿透的问题在多服务、多限界上下文的场景下异步事件是刚需。Python的asyncio能让你用async/await很自然地处理跨上下文事件。但这里有个非常重要的细节事件订阅者的失败不能影响事件发布者的事务所以发布之后订阅者应该尽量做到可重试、幂等的要求。PostgreSQLRedis或者事务性outbox都是不错的选择但那是另一个更深的话题。在单体阶段我建议只需要事务提交后发布事件 事件处理器注册表就够了。真实项目里我最常写的一个事件处理流程是这样的async def on_order_placed(event: OrderPlaced): async with httpx.AsyncClient() as client: await client.post(INVENTORY_SERVICE_URL, json{order_id: event.order_id})这里的要点不是说必须用异步而是必须强调事件处理与原始请求解耦。如果同步一直用requests直接调用一旦下游慢请求线程就会被拖垮最终演化出全部系统罪名的超时长尾问题。6. 依赖注入不用框架Python照样写得干干净净6.1 从手动装配到容器再到手动装配Java系引进了Spring的重量级依赖注入容器后很多Python团队也试图套一套。我试过几个轻量容器如dependency-injector、Kink结论是除非你的项目真的很大且团队有足够纪律否则手动装配反而更透明、更好调试。手动装配在Python里可以写得很优雅。比如在应用入口处建立一个compose_root函数def compose() - AppServices: session_factory create_session_factory(settings.database_url) uow UnitOfWork(session_factory) event_bus InProcessEventBus() order_app OrderApplicationService(uow, event_bus) return AppServices(order_app)这一处就是你的依赖容器的配置单所有依赖一目了然。要换测试替身就改这个地方要换数据库实现也改这个地方。你永远不需要去猜某个全局单例是谁注入的、谁替换的。6.2 Protocol和ABC怎么选在领域层的抽象里我推荐用typing.Protocol。它在运行时是鸭子类型不强制要求实现类显式继承这在写测试替身时非常舒服。你只用做from typing import Protocol class OrderRepository(Protocol): def get(self, order_id: OrderId) - Order: ... def save(self, order: Order) - None: ...测试时写一个FakeOrderRepository满足同样的方法签名就能用不需要它继承任何东西。这让领域层保持零框架依赖也让你的IDE和mypy都能做很好的类型检查。唯一要注意的是如果团队里有人对Protocol不够熟很容易漏掉某个方法导致运行时AttributeError所以建议配合严格静态检查或者强一点的lint规则在CI阶段就抓出来。6.3 应用服务、领域服务、基础设施服务层与层之间怎么摆很多Python项目的结构是扁平的一堆service但我推荐在模块层面明确划分层级。下面这段是我反复校验过的一种目录结构app/ application/ # 应用服务、DTO、命令/查询处理 domain/ # 实体、值对象、聚合根、领域服务、领域事件、Repository接口 infrastructure/ # 数据库实现、缓存、外部API客户端、消息队列适配 presentation/ # HTTP接口、CLI、事件监听的外层入口 composition.py # 手动装配根注意presentation里放接口和路由但它不能直接触碰领域对象的内部细节。它应该把HTTP请求转成命令对象再交给应用服务。如果让presentation直接调用ORM查询并渲染就毁掉了分层的最后一道防线。7. 复杂业务中的实用技巧从充血贫血到领域服务的取舍7.1 贫血模型还是充血模型别二极管看规则内聚贫血模型实体只含getter/setter逻辑全在外面往往来自被ORM绑架的设计但我也见过只要把所有方法全塞进实体结果实体变成几百行巨型类那同样灾难。判断标准非常简单这项业务规则需不需要两个以上的内部字段协作需要强一致性吗如果只需要拿到两个字段算个值放领域服务里未必更差。比如计算订单税额如果只是把price乘以税率再累加放领域服务OK但如果你想保证免税商品不参与计算或折扣和税不能同时叠加这样的规则而且和订单明细的状态耦合在一起我会倾向于放到订单实体的方法里。这不是玄学是按数据归属来判断。数据长在哪、不可变规则需要哪些数据哪里就是归属地。7.2 业务规则复杂时领域服务怎么和聚合根配合聚合根解决内部一致性问题领域服务解决跨聚合协作问题。拿一个常见场景举例一个客户升级为VIP同时要给他一张优惠券并发送短信通知。客户升级这是customer聚合该干的事发优惠券是campaign聚合干的发短信通知属于外部通信。把这三件事都写进Customer.upgrade()这个聚合根就会背上太多不属于它的责任。正确姿势是写一个领域服务VipUpgradeServiceclass VipUpgradeService: def __init__(self, customer_repo, campaign_repo): ... def execute(self, customer_id: str) - None: customer self.customer_repo.get(customer_id) customer.upgrade_vip() self.campaign_repo.grant_vip_coupon(customer_id) # 领域事件里挂上 VipUpgraded监听者负责短信这里customer.upgrade_vip()只改变状态并产生领域事件而服务负责协调多个聚合、仓储和事件的发布。领域服务没有一堆基础设施方法但它本身就是领域逻辑的协同者。这个技巧能让你保持聚合根小巧、领域服务贴近真实业务。7.3 大型复杂领域里的防腐层对上游不给力的模块说不生产级系统绕不开集成旧系统或供应商API。防腐层Anti-Corruption Layer, ACL的职责是翻译把外部模型的脏接口转成我们领域里的干净语言。Python里实现ACL非常简单在infrastructure里建一个adapter目录用一个类封装所有外部调用。比如对接一个老旧的库存系统class LegacyInventoryClient: def query_available_quantity(self, sku: str) - int: response requests.post(LEGACY_URL, json{method: getStock, code: sku}) return response.json()[data][stockNum]你绝不允许外部系统的返回值概念泄漏进domain层。ACL做翻译、做异常转换、做限流重试甚至可以做字段名映射。这套东西在Python里很轻灵但很值钱。8. 迁移与重构路径从一堆事务脚本到领域模型的渐进式重构8.1 识别逻辑腹腔哪些类是差味道发的集中营没人是天生DDD高手绝大多数Python项目都是从Controller/Service里写满if-else开始的。我的重构第一步不是搞目录结构而是找出那些已经明显变态的上帝函数和服务。找它们的信号一个Service方法超过100行而且里面要在5张表之间挑选数据实体对象经常处于非法状态——你经常new一个对象然后手动设字段任其短暂处于违背业务规则的值同一个业务动作分散在多个方法里而且总有忘调的情况测试主要靠mock数据库模拟而不是按业务行为描述找到了就把里面最核心的一两个操作抠出来。比如把一个create_order_and_apply_discount函数里的折扣规则抽到Order.apply_discount()方法里然后写测试覆盖。一次只动一小步别搞big-bang重构。8.2 三步走从草台班子到领域分层建议按三个阶段渐进迁移**阶段一先整理聚合后整理Repository。**把订单、客户这类核心实体抽出来给它们加上业务方法把散落在Service里的操作搬进去。这段时间ORM可以保持原有用法不算最优但可运行。**阶段二引入RepositoryProtocol和UnitOfWork。**把数据库查询从Service层里收走封装到仓储实现里。阶段一丢下的session操作全部移到infrastructure层。期间需要大量测试来保证行为不变。**阶段三拆限界上下文、引入领域事件。**当聚合和方法看起来足够稳定时再把不同业务线按上下文切割在上下文之间引入事件驱动。这一步可以开始思考要不要拆服务、要不要上消息队列。每一步都要保证应用不暂停、测试不回归。这样你说服团队做DDD时的阻力会小很多——没人愿意为了架构洁净去冒一次一次性上线的风险但大家都愿意每周有一条若干分钟的安全迁移。8.3 测试策略围绕领域行为写测试而不是围绕数据库生产级DDD的另一半是测试。我发现相对靠谱的测试金字塔是大量领域单元测试 少量应用集成测试 几个端到端冒烟测试。领域测试直接针对聚合方法不需要数据库、不需要网络、不需要时间服务所有依赖用Fake实现。比如测试订单过期后不能修改明细直接调用Order.expire()然后断言add_line抛异常。应用集成测试则用真实数据库或SQLite内存数据库验证UoW的提交/回滚逻辑。EventBus这类可以注入到FakeEventBus里加一个自定义断言列表确认某个事件是否被正确发布。这一层测试的数量不多因为每个用例都是跨模块链路验证。端到端层面我只写几条主路径冒烟用例确保HTTP入口、应用服务、领域逻辑、仓储、数据库链条没有配置错误。生产级实践里我会专门对领域事件加一个事件一致性测试用来验证事务提交后事件订阅者的幂等性——这种东西不是一次性的而是你以后重构基础设施时特别有效的安全网。9. 尾声这是我对DDDPython长期磨合后的最大心得我最后最想说的是DDD不是银弹也不是被吹捧的PPT概念它是一套关于业务复杂度治理的思维工具而Python因为你手头的表达能力很强非常适合低成本、大步快跑地把这套思维落地。但我更想说的是技术选型永远服务于组织和业务演进——如果你的团队沉淀不出业务语言、你的老板也不想花时间搞领域建模那你起码可以做到一件不亏的事把业务规则从森林一样的Service里搬回领域类里让核心逻辑变得可解释、可测试。就这样一件小事带来的维护收益往往比任何框架和模式都大。等你有一天真正需要DDD的时候你已经站在它门口了。