拓冰网站定制 
1. 项目概述为什么要在 Python 里折腾 DDD用 Python 落地 DDD领域驱动设计这个话题我前后折腾过好几轮。刚接触 DDD 时我也和不少人一样觉得它就是“分层架构 一堆抽象类”的代名词照着 Java 那套模板硬套结果代码越写越重业务团队根本不愿意碰。后来沉下心复盘才发现DDD 的核心从来不是某张代码结构图而是如何让业务复杂度在一个可演进的模型里被安放好。这篇文章我会从概念翻译、工程骨架、核心代码、实操链路和踩坑实录五个方面完整给你一套能在 Python 项目里落地的 DDD 方案。先说清楚这套东西适合谁。如果你的项目里 Service 层已经膨胀到动不动上千行业务规则散落在视图函数和工具函数里每次改需求都要同时改四五个地方那 DDD 的聚合、限界上下文、应用服务这些概念就非常值得你认真过一遍。反过来如果你的业务就是简单的增删改查没有复杂的规则和生命周期那 DDD 反而是负担。它是重型武器不是标配。任何架构思想落到具体语言上都必须经历一次“翻译”。Java 里的接口、抽象类、注解驱动在 Python 里有完全不同的表达方式。Python 的动态特性、dataclass、typing.Protocol甚至contextmanager都能让 DDD 落地得更轻巧但同时也更容易被滥用。这篇文章就是我从“概念正确”到“代码能跑”的完整记录你照着搭骨架再替换成自己的业务大概率能少走很多弯路。1.1 先泼一盆冷水DDD 不是什么项目都该上很多人一听到 DDD第一反应是“我们要不要引入领域驱动设计”仿佛它是一个可以随时安装的插件。实际上 DDD 是一种分析问题的方式它逼迫你去思考这个业务的核心逻辑到底是什么哪些对象拥有自己的生命周期哪些对象只是描述性的没有这种思考你把仓库、实体、值对象全部照搬进代码得到的只是一堆和业务无关的空壳。我见过最典型的案例是一个后台管理项目所有表都是单表增删改查团队依然按照 DDD 的模板拆了七八个包每个实体都带一个抽象基类结果一个改字段的需求要动接口层、应用层、领域层、基础设施层四处代码。这不是 DDD 的错是没有判断复杂度就盲目上架构的错。DDD 真正的用武之地是那些拥有复杂业务规则、多个团队协作、逻辑需要长期演化的系统。比如订单系统、支付系统、风控系统、供应链系统这些领域的核心逻辑一旦散落在各处后期改造成本会指数级上升。所以你在决定落地 DDD 之前先问三个问题业务规则是否足够复杂团队是否有意愿维护清晰的边界业务是否会长线演进如果三个答案都是否定那不如老实写分层架构把 Service 层拆细一点就够了。判断标准不能含糊DDD 是投资不是消费它要在未来几个月甚至几年里通过可控的变更来回报你。1.2 我见过最典型的 Python DDD 翻车现场翻车方式五花八门但根子基本都一样。最常见的是把 Java 的 DDD 模板直接翻译成 Python每个实体继承一个抽象的BaseEntity每个应用服务配一个接口加一个实现类AbstractRepository下面再挂一堆子类。Python 里这么写除了让 IDE 的类层级树变得深不见底没有任何实际价值。第二个翻车点是把仓储接口做成 DAO 的换皮。很多项目里所谓的OrderRepository方法列表是insert、update、delete、select_by_id这和OrderDAO没有任何区别。真正的仓储是以聚合为单位进行持久化的它的目的是让领域层像操作内存对象一样操作领域模型而不是把表结构的访问方式暴露给上层。第三个翻车点更隐蔽就是领域层偷偷依赖了框架。可能一开始只是图方便在实体类里 import 了sqlalchemy或者django.db到后面整个领域模型都被 ORM 的字段类型和懒加载机制绑架。结果业务逻辑没法单独测试升级框架时领域层跟着一起地震。这些问题在后面章节我会逐个给出具体解法现在你先记住一条总原则领域层必须是整个项目里最干净、最不依赖外物的那一层。2. 概念翻译把 DDD 的核心概念映射到 Python 生态DDD 这套理论里有几个核心概念几乎在所有文章里都会出现实体、值对象、聚合、领域服务、应用服务、仓储、限界上下文。这些概念本身不复杂但在 Python 里落地时你需要结合语言特性做取舍。下面我会逐个翻译并附上直接能用的代码风格。2.1 实体、值对象、聚合先分清这三样实体和值对象的区分是 DDD 里最基础也是最重要的判断。判断标准很简单有没有独立的生命周期和身份标识。订单有订单号它的状态会从一个流转到另一个所以是实体金额就是 100 元和 50 元的差别两个 100 元没有任何身份差异所以是值对象。用生活类比就是你是你有身份证号是实体你手里的那张 100 元纸币换一张你还是你钱还是那个钱所以钱在你的交易里只是值对象。在 Python 里我强烈建议用dataclass定义值对象并且把frozenTrue打开让它不可变。这样做有几个收益不可变对象天然可哈希可以直接放进set或者作为字典的 key它也不容易被业务代码意外篡改。比如金额from dataclasses import dataclass from decimal import Decimal dataclass(frozenTrue) class Money: amount: Decimal currency: str CNY def __post_init__(self) - None: if self.amount 0: raise ValueError(金额不能为负数) def add(self, other: Money) - Money: if self.currency ! other.currency: raise ValueError(币种不一致) return Money(amountself.amount other.amount, currencyself.currency)实体的核心是身份标识Python 里可以用dataclass的eqFalse来控制相等性比较from dataclasses import dataclass, field from datetime import datetime from uuid import uuid4 dataclass(eqFalse) class Order: order_id: str field(default_factorylambda: uuid4().hex) status: str pending lines: list[OrderLine] field(default_factorylist) created_at: datetime field(default_factorydatetime.utcnow)聚合是 DDD 里最有价值也最难把握的概念。一个聚合是一组内聚的实体和值对象的集合聚合根是这个集合对外访问的唯一入口。以订单为例Order是聚合根OrderLine是聚合内部的实体除了通过Order你没有办法直接修改某个订单行。这个约束保证了复杂的业务不变量在事务边界内始终成立。聚合边界的划分原则很直白一个事务内要保证强一致的对象放在同一个聚合里。不要把“订单”和“用户”硬塞进同一个聚合他们的变更频率和生命周期都不一样强行绑定只会导致锁竞争和跨聚合污染。2.2 限界上下文与模块划分工程结构的第一刀限界上下文是 DDD 里战略层面的核心概念。简单说就是同一个词在不同业务场景下可能有完全不同的含义和生命周期。比如“商品”在销售上下文里是展示价格和库存的商品在物流上下文里是带重量的货物在财务上下文里是计算成本的存货。强行用一个Product模型覆盖所有场景就是大型业务系统后期腐烂的起点。落到 Python 工程里你应该按业务模块分包而不是按技术分包。很多传统包结构是models/、services/、api/、repositories/但 DDD 建议你按“订单上下文”“库存上下文”“用户上下文”这样的业务边界去划分。每个上下文内部再按分层组织这样当你要剥离某个业务模块时边界是物理存在的。一个可以参考的结构your_project/ ├── config/ ├── domain/ │ ├── order/ │ │ ├── entities.py │ │ ├── value_objects.py │ │ ├── aggregates.py │ │ ├── domain_services.py │ │ └── repository.py │ └── inventory/ │ ├── entities.py │ ├── value_objects.py │ ├── aggregates.py │ └── repository.py ├── application/ │ ├── order/ │ │ ├── use_cases.py │ │ ├── commands.py │ │ └── dto.py ├── infrastructure/ │ ├── db/ │ ├── repositories/ │ └── adapters/ └── interfaces/ ├── api/ └── consumers/注意domain里的repository.py放的是接口不是实现。infrastructure/repositories里才是 SQLAlchemy 或者 Django ORM 的实现。这个结构不是摆设它能保证你的领域代码只依赖抽象不依赖具体技术选型。以后换数据库也好、换缓存也好领域层完全不需要动。2.3 接口抽象与依赖倒置优先用 Protocol 而不是继承抽象类在 Java 里定义仓储接口你可能会写一个interface OrderRepository { Order findById(String id); void save(Order order); }然后用OrderRepositoryImpl去实现。Python 里我推荐用typing.Protocol来做这件事它更符合 Python 的鸭子类型哲学而且不需要让实现类继承任何东西。你只需要定义一个协议类标注方法的签名和类型实现类只要“长得一样”就会被类型检查器认为是合法的。from typing import Protocol from uuid import UUID from domain.order.aggregates import Order class OrderRepository(Protocol): def get_by_id(self, order_id: UUID) - Order | None: ... def save(self, order: Order) - None: ...为什么不用ABC抽象基类强制实现类继承它属于结构上的强耦合而 Protocol 只需要行为匹配测试时随便写一个假对象就能满足类型约束。配合 FastAPI 的Depends或者手动依赖注入你可以非常轻松地在测试环境换成InMemoryOrderRepository。这种“约定优于继承”的风格用起来比 Java 那套舒服很多也更符合 Python 的社区习惯。3. 代码落地一个订单聚合 一个用例 一个仓储的完整骨架概念讲再多不落到代码都是空谈。这一节我会用一个真实感较强的“订单 库存”场景把领域层、应用层、基础设施层各写一部分让你可以直接套用。先说明这里的代码刻意简化了跨上下文的事件通信但结构上保持了 DDD 的完整性。3.1 领域层用 dataclass 写出可测试的实体与聚合根先定义值对象和实体。订单行OrderLine在这个聚合里没有全局身份它依附于订单存在所以是实体但它的相等性比较可以用行号来做from dataclasses import dataclass, field from decimal import Decimal from uuid import UUID, uuid4 from datetime import datetime from domain.order.value_objects import Money, OrderStatus dataclass(eqFalse) class OrderLine: line_no: int product_id: UUID quantity: int price: Money def total(self) - Money: return Money(amountself.price.amount * self.quantity, currencyself.price.currency) dataclass(eqFalse) class Order: order_id: UUID field(default_factoryuuid4) customer_id: UUID field(default_factoryuuid4) status: OrderStatus OrderStatus.PENDING lines: list[OrderLine] field(default_factorylist) created_at: datetime field(default_factorydatetime.utcnow) version: int 0 def add_line(self, product_id: UUID, quantity: int, price: Money) - None: if self.status ! OrderStatus.PENDING: raise ValueError(只有待支付的订单才能加商品) if quantity 0: raise ValueError(商品数量必须为正) line_no len(self.lines) 1 self.lines.append(OrderLine(line_noline_no, product_idproduct_id, quantityquantity, priceprice)) def total(self) - Money: return sum((line.total() for line in self.lines), Money(Decimal(0), CNY)) def place(self) - list[DomainEvent]: if not self.lines: raise ValueError(不能下单空订单) if self.status ! OrderStatus.PENDING: raise ValueError(订单状态不允许重复提交) self.status OrderStatus.PLACED self.version 1 return [OrderPlaced(order_idself.order_id, customer_idself.customer_id, totalself.total())]这段代码有几个关键点。第一所有业务规则都写在实体方法内部加商品前检查状态下单前检查订单非空、状态合法。这是与“贫血模型”最本质的区别也是 DDD 的初衷。第二每次领域行为触发状态变更后返回一个领域事件列表。这样做的好处是应用服务可以在事务提交成功后再发布事件避免“领域事件发了但事务没提交”的经典脏读问题。3.2 仓储抽象与基础设施实现方向别搞反仓储接口放在领域层实现在基础设施层依赖方向是从外向内。领域层完全不认识 SQLAlchemy它只需要get_by_id和save两个方法。实现类可以长这样from uuid import UUID from sqlalchemy.orm import Session from domain.order.aggregates import Order from infrastructure.db import OrderModel, OrderLineModel from infrastructure.repositories import OrderRepository class SqlAlchemyOrderRepository(OrderRepository): def __init__(self, session: Session) - None: self.session session def get_by_id(self, order_id: UUID) - Order | None: order_model self.session.get(OrderModel, str(order_id)) if order_model is None: return None return self._to_domain(order_model) def save(self, order: Order) - None: order_model self._to_model(order) self.session.merge(order_model) def _to_domain(self, order_model: OrderModel) - Order: lines [ OrderLine(line_noline.line_no, product_idUUID(line.product_id), quantityline.quantity, priceMoney(Decimal(line.amount), line.currency)) for line in order_model.lines ] return Order(order_idUUID(order_model.order_id), customer_idUUID(order_model.customer_id), statusOrderStatus(order_model.status), lineslines, versionorder_model.version) def _to_model(self, order: Order) - OrderModel: return OrderModel( order_idstr(order.order_id), customer_idstr(order.customer_id), statusorder.status.value, versionorder.version, lines[ OrderLineModel(line_noline.line_no, product_idstr(line.product_id), quantityline.quantity, amountstr(line.price.amount), currencyline.price.currency) for line in order.lines ], )请注意这里的仓储并不是简单的表映射它承载了“聚合持久化”的职责。加载订单时把OrderLine一并恢复出来保存时用merge把聚合整体写回。这要求聚合里的内容不能是懒加载的所以在领域层用普通的dataclass列表而不是 ORM 的关系属性。3.3 应用层与工作单元用例是“编排”出来的不是长出来的应用服务Application Service是领域层和外部世界的翻译官。它不承载业务规则只负责任务编排拿到输入、调用领域模型、把结果持久化、发布事件。这里必须引入工作单元Unit of Work否则多个仓储一起操作时根本没有事务边界。我惯用的方式是用上下文管理器from contextlib import contextmanager from collections.abc import Iterator from sqlalchemy.orm import Session class UnitOfWork: def __init__(self, session_factory) - None: self.session_factory session_factory contextmanager def __enter__(self) - Iterator[Session]: self.session self.session_factory() try: yield self.session self.session.commit() except Exception: self.session.rollback() raise finally: self.session.close()应用层的用例类拿到这个UnitOfWork就可以在单个事务里轻松操作多个仓储from dataclasses import dataclass from uuid import UUID from domain.order.aggregates import Order from domain.order.events import OrderPlaced from infrastructure.repositories import OrderRepository from application.order.unit_of_work import UnitOfWork dataclass class PlaceOrderCommand: customer_id: UUID lines: list[dict] class PlaceOrderUseCase: def __init__(self, uow: UnitOfWork, order_repo: OrderRepository, event_publisher) - None: self.uow uow self.order_repo order_repo self.event_publisher event_publisher def execute(self, cmd: PlaceOrderCommand) - UUID: order Order(customer_idcmd.customer_id) for line in cmd.lines: order.add_line(product_idline[product_id], quantityline[quantity], priceMoney(Decimal(line[price]), CNY)) events order.place() with self.uow as session: self.order_repo.save(order) published events # 事务已提交可以安全发布 for event in published: self.event_publisher.publish(event) return order.order_id这段代码把 DDD 里最容易被搞混的两个东西分得很清楚Order与PlaceOrderUseCase。领域层的Order只知道自己的规则应用层的用例只知道“调谁、存谁、发什么事件”。以后业务规则变了改实体流程变了改用例。两个维度的变化互不干扰。3.4 领域事件跨聚合解耦的一个轻量方案领域事件是 DDD 里处理跨聚合一致性的重要手段。比如下单后要扣减库存你不需要在同一个事务里同时锁订单表和库存表而是订单下单成功后发出OrderPlaced事件库存上下文监听这个事件做对应的扣减。这属于“最终一致性”的范畴但能有效缩小事务锁的范围提升并发能力。事件本身就是一个普通的 dataclassfrom dataclasses import dataclass from uuid import UUID from domain.order.value_objects import Money dataclass(frozenTrue) class OrderPlaced: order_id: UUID customer_id: UUID total: Money在系统不复杂的时候一个简单的内存事件总线就够用了from collections.abc import Callable from typing import Any class SimpleEventBus: def __init__(self) - None: self._subscribers: dict[type, list[Callable]] {} def subscribe(self, event_type: type, handler: Callable) - None: self._subscribers.setdefault(event_type, []).append(handler) def publish(self, event: Any) - None: for handler in self._subscribers.get(type(event), []): handler(event)等系统演进到需要跨进程通信时再把这个总线换成消息队列的实现。领域层只需要依赖事件对象本身不需要关心它是怎么被发布的。这也是 DDD 的魅力技术细节永远处在可替换的位置上。4. 实操过程一次完整的“下单”用例全链路走查前面给的代码都是零件这一节把它们组装起来模拟一次真实的 HTTP 请求从进来到落库的全过程。我带你从接口层走一遍顺便说一下各层之间的依赖关系是怎么被组装出来的。4.1 从 FastAPI 接口到应用层一条清晰的调用链接口层只负责协议转换接收请求体、解析参数、调用应用服务、序列化返回结果。它不知道订单的业务规则是什么也不知道数据存在哪个数据库。用 FastAPI 可以这样写from uuid import UUID from fastapi import APIRouter, Depends from application.order.use_cases import PlaceOrderUseCase, PlaceOrderCommand from infrastructure.repositories import SqlAlchemyOrderRepository from application.order.unit_of_work import UnitOfWork from interfaces.api.dependencies import get_session_factory router APIRouter() def get_place_order_uc() - PlaceOrderUseCase: uow UnitOfWork(get_session_factory()) repo SqlAlchemyOrderRepository(sessionuow.session_factory()) return PlaceOrderUseCase(uowuow, order_reporepo, event_publisherevent_bus) router.post(/orders) def place_order(payload: dict, uc: PlaceOrderUseCase Depends(get_place_order_uc)): cmd PlaceOrderCommand(customer_idUUID(payload[customer_id]), linespayload[lines]) order_id uc.execute(cmd) return {order_id: str(order_id)}这里有一个细节值得注意get_place_order_uc看起来像是在“手动注入”但它比把 ORM 的 session 直接透传到接口层要干净得多。真正的依赖关系是接口层依赖应用层接口应用层依赖领域层接口领域层不依赖任何一层。以后如果换了数据库只需要修改依赖组装处接口和应用层代码一行都不动。完整的调用链大概是这样的HTTP 请求进来 → FastAPI 调用PlaceOrderUseCase.execute→ 用例调用Order的add_line和place方法 → 领域事件OrderPlaced被返回 → 用例用UnitOfWork提交订单 → 事件发布到SimpleEventBus→ 库存监听器扣减库存。整个过程领域层从头到尾没有碰过网络、数据库和消息队列。4.2 事务边界与并发保护DDD 落地最后一块拼图很多团队尝试 DDD 后最大的困惑是事务到底应该放在哪一层我的答案很明确事务边界放在应用层但事务的内容由仓储和 UnitOfWork 共同管理。刚才的代码里with self.uow as session就是整个用例的事务边界它保证order_repo.save和后续可能发生的其他仓储写入要么全部成功要么全部回滚。但事务只能解决一致性问题解决不了并发问题。两个请求同时提交同一个订单可能会互相覆盖状态。所以我通常会做几个数据库层面的补充第一在订单表上给order_id加唯一约束第二为聚合根加version字段实现乐观锁第三如果确实需要实时一致性可以在更新时加上条件WHERE version :old_version。乐观锁的具体 SQL 可以这样UPDATE orders SET status :new_status, version version 1 WHERE order_id :order_id AND version :old_version如果影响行数为 0说明版本冲突直接返回“操作过于频繁请重试”。领域层只感知到这次保存失败不需要知道锁的实现细节。4.3 测试策略让领域层不装数据库也能稳稳地跑DDD 一个很直接的收益就是测试能按层级清晰分层。领域层的测试完全不碰数据库用内存仓储就能跑通整个用例。先写一个假的仓储from uuid import UUID from domain.order.aggregates import Order class InMemoryOrderRepo: def __init__(self) - None: self._orders: dict[UUID, Order] {} def get_by_id(self, order_id: UUID) - Order | None: return self._orders.get(order_id) def save(self, order: Order) - None: self._orders[order.order_id] order然后领域测试可以这样写def test_order_place_updates_status() - None: order Order() order.add_line(product_iduuid4(), quantity1, priceMoney(Decimal(100.00), CNY)) events order.place() assert order.status OrderStatus.PLACED assert len(events) 1 assert isinstance(events[0], OrderPlaced)应用层测试则使用真实的 UseCase 内存仓储 内存事件总线不依赖真实的数据库。基础设施层的集成测试留一小部分给 CI 去跑真实数据库。这样既保证了领域逻辑的可靠又不至于让测试跑得很慢。实战下来这个测试金字塔在 Python 项目里非常好用。5. 常见问题与排查技巧实录最后这一章我要把真实项目里最常踩的坑集中盘点一遍。这些问题你在网上搜 DDD 时很少有人系统性地提但几乎每个团队都会遇到。5.1 问题速查表问题现象根因解决思路聚合根越来越大一个类几百行加一个字段要改十个方法边界划错把不相关的对象塞进了同一个聚合回到业务分析看哪些对象生命周期绑定哪些只是临时关联仓储接口方法列表变成insert/update/delete把 DAO 当仓储用丢失了聚合概念重设仓储语义改成以聚合为单位的方法如get_by_id/save订单保存后领域事件总报“事件先发了但库里没有”在事务提交前发布事件移动到应用层事务提交之后再 publish领域层里出现 SQLAlchemy 的 Session 类型依赖倒置没做好领域被基础设施污染用 Protocol 抽象仓储基础设施实现放外层Service 层几千行却号称用了 DDD贫血模型实体上只有 getter/setter把业务方法按行为搬到实体里服务层只做编排测试环境依赖真实数据库跑一次要三分钟没有内存仓储和测试替身用 InMemory repo fake event bus 搭建测试替身5.2 Python 的类型提示把“自由”变成“约束”动态语言落地 DDD 有一个天然劣势没有编译器帮你检查接口是否被正确实现。因此类型提示是你唯一的便宜安全保障。我在项目里会强制使用mypy或pyright做全量检查并且把仓库接口统一用Protocol定义。这样任何实现了OrderRepository协议的类如果方法签名不匹配CI 阶段直接报错。IDE 也能在你调用repo.save(order)时自动补全和提示函数签名开发效率并不输给静态语言。另外一个容易被忽略的点是Python 的dataclass在生成__init__时默认允许可变对象做字段这会让值对象的“不可变性”形同虚设。所以我在做 DDD 时坚持使用frozenTrue并且对decimal.Decimal、datetime这类不可变类型做__post_init__校验。这个习惯能避免很多隐蔽的 bug。5.3 渐进式重构建议别把老系统推倒重来最后想和你说一句真心话不要因为看了这篇文章就回去把线上系统全盘重写。DDD 落地的最佳方式是在一个边界清晰的新模块里先试点。比如老系统里有一个订单导出的功能长期被改来改去你可以先把它圈出来在domain/order/里建出聚合和用例再让接口层切换过来。等团队真的体会到“改一个业务规则只动一个地方”的爽感之后再逐步扩大范围。架构演进慢一点损失会小很多。我在实际项目里还有一个习惯每改一个需求会顺手问自己“这个改动落在哪个层领域模型需要动吗”如果答案经常是“领域模型不动只改用例”那说明边界划对了如果每次都要同时改动实体、仓储、接口那大概率是聚合边界或者上下文划分出了问题。回到业务本身去重新分析永远比埋头改代码更靠谱。 Python实践DDD：聚合、仓储与领域事件的完整落地指南
1. 项目概述为什么要在 Python 里折腾 DDD用 Python 落地 DDD领域驱动设计这个话题我前后折腾过好几轮。刚接触 DDD 时我也和不少人一样觉得它就是“分层架构 一堆抽象类”的代名词照着 Java 那套模板硬套结果代码越写越重业务团队根本不愿意碰。后来沉下心复盘才发现DDD 的核心从来不是某张代码结构图而是如何让业务复杂度在一个可演进的模型里被安放好。这篇文章我会从概念翻译、工程骨架、核心代码、实操链路和踩坑实录五个方面完整给你一套能在 Python 项目里落地的 DDD 方案。先说清楚这套东西适合谁。如果你的项目里 Service 层已经膨胀到动不动上千行业务规则散落在视图函数和工具函数里每次改需求都要同时改四五个地方那 DDD 的聚合、限界上下文、应用服务这些概念就非常值得你认真过一遍。反过来如果你的业务就是简单的增删改查没有复杂的规则和生命周期那 DDD 反而是负担。它是重型武器不是标配。任何架构思想落到具体语言上都必须经历一次“翻译”。Java 里的接口、抽象类、注解驱动在 Python 里有完全不同的表达方式。Python 的动态特性、dataclass、typing.Protocol甚至contextmanager都能让 DDD 落地得更轻巧但同时也更容易被滥用。这篇文章就是我从“概念正确”到“代码能跑”的完整记录你照着搭骨架再替换成自己的业务大概率能少走很多弯路。1.1 先泼一盆冷水DDD 不是什么项目都该上很多人一听到 DDD第一反应是“我们要不要引入领域驱动设计”仿佛它是一个可以随时安装的插件。实际上 DDD 是一种分析问题的方式它逼迫你去思考这个业务的核心逻辑到底是什么哪些对象拥有自己的生命周期哪些对象只是描述性的没有这种思考你把仓库、实体、值对象全部照搬进代码得到的只是一堆和业务无关的空壳。我见过最典型的案例是一个后台管理项目所有表都是单表增删改查团队依然按照 DDD 的模板拆了七八个包每个实体都带一个抽象基类结果一个改字段的需求要动接口层、应用层、领域层、基础设施层四处代码。这不是 DDD 的错是没有判断复杂度就盲目上架构的错。DDD 真正的用武之地是那些拥有复杂业务规则、多个团队协作、逻辑需要长期演化的系统。比如订单系统、支付系统、风控系统、供应链系统这些领域的核心逻辑一旦散落在各处后期改造成本会指数级上升。所以你在决定落地 DDD 之前先问三个问题业务规则是否足够复杂团队是否有意愿维护清晰的边界业务是否会长线演进如果三个答案都是否定那不如老实写分层架构把 Service 层拆细一点就够了。判断标准不能含糊DDD 是投资不是消费它要在未来几个月甚至几年里通过可控的变更来回报你。1.2 我见过最典型的 Python DDD 翻车现场翻车方式五花八门但根子基本都一样。最常见的是把 Java 的 DDD 模板直接翻译成 Python每个实体继承一个抽象的BaseEntity每个应用服务配一个接口加一个实现类AbstractRepository下面再挂一堆子类。Python 里这么写除了让 IDE 的类层级树变得深不见底没有任何实际价值。第二个翻车点是把仓储接口做成 DAO 的换皮。很多项目里所谓的OrderRepository方法列表是insert、update、delete、select_by_id这和OrderDAO没有任何区别。真正的仓储是以聚合为单位进行持久化的它的目的是让领域层像操作内存对象一样操作领域模型而不是把表结构的访问方式暴露给上层。第三个翻车点更隐蔽就是领域层偷偷依赖了框架。可能一开始只是图方便在实体类里 import 了sqlalchemy或者django.db到后面整个领域模型都被 ORM 的字段类型和懒加载机制绑架。结果业务逻辑没法单独测试升级框架时领域层跟着一起地震。这些问题在后面章节我会逐个给出具体解法现在你先记住一条总原则领域层必须是整个项目里最干净、最不依赖外物的那一层。2. 概念翻译把 DDD 的核心概念映射到 Python 生态DDD 这套理论里有几个核心概念几乎在所有文章里都会出现实体、值对象、聚合、领域服务、应用服务、仓储、限界上下文。这些概念本身不复杂但在 Python 里落地时你需要结合语言特性做取舍。下面我会逐个翻译并附上直接能用的代码风格。2.1 实体、值对象、聚合先分清这三样实体和值对象的区分是 DDD 里最基础也是最重要的判断。判断标准很简单有没有独立的生命周期和身份标识。订单有订单号它的状态会从一个流转到另一个所以是实体金额就是 100 元和 50 元的差别两个 100 元没有任何身份差异所以是值对象。用生活类比就是你是你有身份证号是实体你手里的那张 100 元纸币换一张你还是你钱还是那个钱所以钱在你的交易里只是值对象。在 Python 里我强烈建议用dataclass定义值对象并且把frozenTrue打开让它不可变。这样做有几个收益不可变对象天然可哈希可以直接放进set或者作为字典的 key它也不容易被业务代码意外篡改。比如金额from dataclasses import dataclass from decimal import Decimal dataclass(frozenTrue) class Money: amount: Decimal currency: str CNY def __post_init__(self) - None: if self.amount 0: raise ValueError(金额不能为负数) def add(self, other: Money) - Money: if self.currency ! other.currency: raise ValueError(币种不一致) return Money(amountself.amount other.amount, currencyself.currency)实体的核心是身份标识Python 里可以用dataclass的eqFalse来控制相等性比较from dataclasses import dataclass, field from datetime import datetime from uuid import uuid4 dataclass(eqFalse) class Order: order_id: str field(default_factorylambda: uuid4().hex) status: str pending lines: list[OrderLine] field(default_factorylist) created_at: datetime field(default_factorydatetime.utcnow)聚合是 DDD 里最有价值也最难把握的概念。一个聚合是一组内聚的实体和值对象的集合聚合根是这个集合对外访问的唯一入口。以订单为例Order是聚合根OrderLine是聚合内部的实体除了通过Order你没有办法直接修改某个订单行。这个约束保证了复杂的业务不变量在事务边界内始终成立。聚合边界的划分原则很直白一个事务内要保证强一致的对象放在同一个聚合里。不要把“订单”和“用户”硬塞进同一个聚合他们的变更频率和生命周期都不一样强行绑定只会导致锁竞争和跨聚合污染。2.2 限界上下文与模块划分工程结构的第一刀限界上下文是 DDD 里战略层面的核心概念。简单说就是同一个词在不同业务场景下可能有完全不同的含义和生命周期。比如“商品”在销售上下文里是展示价格和库存的商品在物流上下文里是带重量的货物在财务上下文里是计算成本的存货。强行用一个Product模型覆盖所有场景就是大型业务系统后期腐烂的起点。落到 Python 工程里你应该按业务模块分包而不是按技术分包。很多传统包结构是models/、services/、api/、repositories/但 DDD 建议你按“订单上下文”“库存上下文”“用户上下文”这样的业务边界去划分。每个上下文内部再按分层组织这样当你要剥离某个业务模块时边界是物理存在的。一个可以参考的结构your_project/ ├── config/ ├── domain/ │ ├── order/ │ │ ├── entities.py │ │ ├── value_objects.py │ │ ├── aggregates.py │ │ ├── domain_services.py │ │ └── repository.py │ └── inventory/ │ ├── entities.py │ ├── value_objects.py │ ├── aggregates.py │ └── repository.py ├── application/ │ ├── order/ │ │ ├── use_cases.py │ │ ├── commands.py │ │ └── dto.py ├── infrastructure/ │ ├── db/ │ ├── repositories/ │ └── adapters/ └── interfaces/ ├── api/ └── consumers/注意domain里的repository.py放的是接口不是实现。infrastructure/repositories里才是 SQLAlchemy 或者 Django ORM 的实现。这个结构不是摆设它能保证你的领域代码只依赖抽象不依赖具体技术选型。以后换数据库也好、换缓存也好领域层完全不需要动。2.3 接口抽象与依赖倒置优先用 Protocol 而不是继承抽象类在 Java 里定义仓储接口你可能会写一个interface OrderRepository { Order findById(String id); void save(Order order); }然后用OrderRepositoryImpl去实现。Python 里我推荐用typing.Protocol来做这件事它更符合 Python 的鸭子类型哲学而且不需要让实现类继承任何东西。你只需要定义一个协议类标注方法的签名和类型实现类只要“长得一样”就会被类型检查器认为是合法的。from typing import Protocol from uuid import UUID from domain.order.aggregates import Order class OrderRepository(Protocol): def get_by_id(self, order_id: UUID) - Order | None: ... def save(self, order: Order) - None: ...为什么不用ABC抽象基类强制实现类继承它属于结构上的强耦合而 Protocol 只需要行为匹配测试时随便写一个假对象就能满足类型约束。配合 FastAPI 的Depends或者手动依赖注入你可以非常轻松地在测试环境换成InMemoryOrderRepository。这种“约定优于继承”的风格用起来比 Java 那套舒服很多也更符合 Python 的社区习惯。3. 代码落地一个订单聚合 一个用例 一个仓储的完整骨架概念讲再多不落到代码都是空谈。这一节我会用一个真实感较强的“订单 库存”场景把领域层、应用层、基础设施层各写一部分让你可以直接套用。先说明这里的代码刻意简化了跨上下文的事件通信但结构上保持了 DDD 的完整性。3.1 领域层用 dataclass 写出可测试的实体与聚合根先定义值对象和实体。订单行OrderLine在这个聚合里没有全局身份它依附于订单存在所以是实体但它的相等性比较可以用行号来做from dataclasses import dataclass, field from decimal import Decimal from uuid import UUID, uuid4 from datetime import datetime from domain.order.value_objects import Money, OrderStatus dataclass(eqFalse) class OrderLine: line_no: int product_id: UUID quantity: int price: Money def total(self) - Money: return Money(amountself.price.amount * self.quantity, currencyself.price.currency) dataclass(eqFalse) class Order: order_id: UUID field(default_factoryuuid4) customer_id: UUID field(default_factoryuuid4) status: OrderStatus OrderStatus.PENDING lines: list[OrderLine] field(default_factorylist) created_at: datetime field(default_factorydatetime.utcnow) version: int 0 def add_line(self, product_id: UUID, quantity: int, price: Money) - None: if self.status ! OrderStatus.PENDING: raise ValueError(只有待支付的订单才能加商品) if quantity 0: raise ValueError(商品数量必须为正) line_no len(self.lines) 1 self.lines.append(OrderLine(line_noline_no, product_idproduct_id, quantityquantity, priceprice)) def total(self) - Money: return sum((line.total() for line in self.lines), Money(Decimal(0), CNY)) def place(self) - list[DomainEvent]: if not self.lines: raise ValueError(不能下单空订单) if self.status ! OrderStatus.PENDING: raise ValueError(订单状态不允许重复提交) self.status OrderStatus.PLACED self.version 1 return [OrderPlaced(order_idself.order_id, customer_idself.customer_id, totalself.total())]这段代码有几个关键点。第一所有业务规则都写在实体方法内部加商品前检查状态下单前检查订单非空、状态合法。这是与“贫血模型”最本质的区别也是 DDD 的初衷。第二每次领域行为触发状态变更后返回一个领域事件列表。这样做的好处是应用服务可以在事务提交成功后再发布事件避免“领域事件发了但事务没提交”的经典脏读问题。3.2 仓储抽象与基础设施实现方向别搞反仓储接口放在领域层实现在基础设施层依赖方向是从外向内。领域层完全不认识 SQLAlchemy它只需要get_by_id和save两个方法。实现类可以长这样from uuid import UUID from sqlalchemy.orm import Session from domain.order.aggregates import Order from infrastructure.db import OrderModel, OrderLineModel from infrastructure.repositories import OrderRepository class SqlAlchemyOrderRepository(OrderRepository): def __init__(self, session: Session) - None: self.session session def get_by_id(self, order_id: UUID) - Order | None: order_model self.session.get(OrderModel, str(order_id)) if order_model is None: return None return self._to_domain(order_model) def save(self, order: Order) - None: order_model self._to_model(order) self.session.merge(order_model) def _to_domain(self, order_model: OrderModel) - Order: lines [ OrderLine(line_noline.line_no, product_idUUID(line.product_id), quantityline.quantity, priceMoney(Decimal(line.amount), line.currency)) for line in order_model.lines ] return Order(order_idUUID(order_model.order_id), customer_idUUID(order_model.customer_id), statusOrderStatus(order_model.status), lineslines, versionorder_model.version) def _to_model(self, order: Order) - OrderModel: return OrderModel( order_idstr(order.order_id), customer_idstr(order.customer_id), statusorder.status.value, versionorder.version, lines[ OrderLineModel(line_noline.line_no, product_idstr(line.product_id), quantityline.quantity, amountstr(line.price.amount), currencyline.price.currency) for line in order.lines ], )请注意这里的仓储并不是简单的表映射它承载了“聚合持久化”的职责。加载订单时把OrderLine一并恢复出来保存时用merge把聚合整体写回。这要求聚合里的内容不能是懒加载的所以在领域层用普通的dataclass列表而不是 ORM 的关系属性。3.3 应用层与工作单元用例是“编排”出来的不是长出来的应用服务Application Service是领域层和外部世界的翻译官。它不承载业务规则只负责任务编排拿到输入、调用领域模型、把结果持久化、发布事件。这里必须引入工作单元Unit of Work否则多个仓储一起操作时根本没有事务边界。我惯用的方式是用上下文管理器from contextlib import contextmanager from collections.abc import Iterator from sqlalchemy.orm import Session class UnitOfWork: def __init__(self, session_factory) - None: self.session_factory session_factory contextmanager def __enter__(self) - Iterator[Session]: self.session self.session_factory() try: yield self.session self.session.commit() except Exception: self.session.rollback() raise finally: self.session.close()应用层的用例类拿到这个UnitOfWork就可以在单个事务里轻松操作多个仓储from dataclasses import dataclass from uuid import UUID from domain.order.aggregates import Order from domain.order.events import OrderPlaced from infrastructure.repositories import OrderRepository from application.order.unit_of_work import UnitOfWork dataclass class PlaceOrderCommand: customer_id: UUID lines: list[dict] class PlaceOrderUseCase: def __init__(self, uow: UnitOfWork, order_repo: OrderRepository, event_publisher) - None: self.uow uow self.order_repo order_repo self.event_publisher event_publisher def execute(self, cmd: PlaceOrderCommand) - UUID: order Order(customer_idcmd.customer_id) for line in cmd.lines: order.add_line(product_idline[product_id], quantityline[quantity], priceMoney(Decimal(line[price]), CNY)) events order.place() with self.uow as session: self.order_repo.save(order) published events # 事务已提交可以安全发布 for event in published: self.event_publisher.publish(event) return order.order_id这段代码把 DDD 里最容易被搞混的两个东西分得很清楚Order与PlaceOrderUseCase。领域层的Order只知道自己的规则应用层的用例只知道“调谁、存谁、发什么事件”。以后业务规则变了改实体流程变了改用例。两个维度的变化互不干扰。3.4 领域事件跨聚合解耦的一个轻量方案领域事件是 DDD 里处理跨聚合一致性的重要手段。比如下单后要扣减库存你不需要在同一个事务里同时锁订单表和库存表而是订单下单成功后发出OrderPlaced事件库存上下文监听这个事件做对应的扣减。这属于“最终一致性”的范畴但能有效缩小事务锁的范围提升并发能力。事件本身就是一个普通的 dataclassfrom dataclasses import dataclass from uuid import UUID from domain.order.value_objects import Money dataclass(frozenTrue) class OrderPlaced: order_id: UUID customer_id: UUID total: Money在系统不复杂的时候一个简单的内存事件总线就够用了from collections.abc import Callable from typing import Any class SimpleEventBus: def __init__(self) - None: self._subscribers: dict[type, list[Callable]] {} def subscribe(self, event_type: type, handler: Callable) - None: self._subscribers.setdefault(event_type, []).append(handler) def publish(self, event: Any) - None: for handler in self._subscribers.get(type(event), []): handler(event)等系统演进到需要跨进程通信时再把这个总线换成消息队列的实现。领域层只需要依赖事件对象本身不需要关心它是怎么被发布的。这也是 DDD 的魅力技术细节永远处在可替换的位置上。4. 实操过程一次完整的“下单”用例全链路走查前面给的代码都是零件这一节把它们组装起来模拟一次真实的 HTTP 请求从进来到落库的全过程。我带你从接口层走一遍顺便说一下各层之间的依赖关系是怎么被组装出来的。4.1 从 FastAPI 接口到应用层一条清晰的调用链接口层只负责协议转换接收请求体、解析参数、调用应用服务、序列化返回结果。它不知道订单的业务规则是什么也不知道数据存在哪个数据库。用 FastAPI 可以这样写from uuid import UUID from fastapi import APIRouter, Depends from application.order.use_cases import PlaceOrderUseCase, PlaceOrderCommand from infrastructure.repositories import SqlAlchemyOrderRepository from application.order.unit_of_work import UnitOfWork from interfaces.api.dependencies import get_session_factory router APIRouter() def get_place_order_uc() - PlaceOrderUseCase: uow UnitOfWork(get_session_factory()) repo SqlAlchemyOrderRepository(sessionuow.session_factory()) return PlaceOrderUseCase(uowuow, order_reporepo, event_publisherevent_bus) router.post(/orders) def place_order(payload: dict, uc: PlaceOrderUseCase Depends(get_place_order_uc)): cmd PlaceOrderCommand(customer_idUUID(payload[customer_id]), linespayload[lines]) order_id uc.execute(cmd) return {order_id: str(order_id)}这里有一个细节值得注意get_place_order_uc看起来像是在“手动注入”但它比把 ORM 的 session 直接透传到接口层要干净得多。真正的依赖关系是接口层依赖应用层接口应用层依赖领域层接口领域层不依赖任何一层。以后如果换了数据库只需要修改依赖组装处接口和应用层代码一行都不动。完整的调用链大概是这样的HTTP 请求进来 → FastAPI 调用PlaceOrderUseCase.execute→ 用例调用Order的add_line和place方法 → 领域事件OrderPlaced被返回 → 用例用UnitOfWork提交订单 → 事件发布到SimpleEventBus→ 库存监听器扣减库存。整个过程领域层从头到尾没有碰过网络、数据库和消息队列。4.2 事务边界与并发保护DDD 落地最后一块拼图很多团队尝试 DDD 后最大的困惑是事务到底应该放在哪一层我的答案很明确事务边界放在应用层但事务的内容由仓储和 UnitOfWork 共同管理。刚才的代码里with self.uow as session就是整个用例的事务边界它保证order_repo.save和后续可能发生的其他仓储写入要么全部成功要么全部回滚。但事务只能解决一致性问题解决不了并发问题。两个请求同时提交同一个订单可能会互相覆盖状态。所以我通常会做几个数据库层面的补充第一在订单表上给order_id加唯一约束第二为聚合根加version字段实现乐观锁第三如果确实需要实时一致性可以在更新时加上条件WHERE version :old_version。乐观锁的具体 SQL 可以这样UPDATE orders SET status :new_status, version version 1 WHERE order_id :order_id AND version :old_version如果影响行数为 0说明版本冲突直接返回“操作过于频繁请重试”。领域层只感知到这次保存失败不需要知道锁的实现细节。4.3 测试策略让领域层不装数据库也能稳稳地跑DDD 一个很直接的收益就是测试能按层级清晰分层。领域层的测试完全不碰数据库用内存仓储就能跑通整个用例。先写一个假的仓储from uuid import UUID from domain.order.aggregates import Order class InMemoryOrderRepo: def __init__(self) - None: self._orders: dict[UUID, Order] {} def get_by_id(self, order_id: UUID) - Order | None: return self._orders.get(order_id) def save(self, order: Order) - None: self._orders[order.order_id] order然后领域测试可以这样写def test_order_place_updates_status() - None: order Order() order.add_line(product_iduuid4(), quantity1, priceMoney(Decimal(100.00), CNY)) events order.place() assert order.status OrderStatus.PLACED assert len(events) 1 assert isinstance(events[0], OrderPlaced)应用层测试则使用真实的 UseCase 内存仓储 内存事件总线不依赖真实的数据库。基础设施层的集成测试留一小部分给 CI 去跑真实数据库。这样既保证了领域逻辑的可靠又不至于让测试跑得很慢。实战下来这个测试金字塔在 Python 项目里非常好用。5. 常见问题与排查技巧实录最后这一章我要把真实项目里最常踩的坑集中盘点一遍。这些问题你在网上搜 DDD 时很少有人系统性地提但几乎每个团队都会遇到。5.1 问题速查表问题现象根因解决思路聚合根越来越大一个类几百行加一个字段要改十个方法边界划错把不相关的对象塞进了同一个聚合回到业务分析看哪些对象生命周期绑定哪些只是临时关联仓储接口方法列表变成insert/update/delete把 DAO 当仓储用丢失了聚合概念重设仓储语义改成以聚合为单位的方法如get_by_id/save订单保存后领域事件总报“事件先发了但库里没有”在事务提交前发布事件移动到应用层事务提交之后再 publish领域层里出现 SQLAlchemy 的 Session 类型依赖倒置没做好领域被基础设施污染用 Protocol 抽象仓储基础设施实现放外层Service 层几千行却号称用了 DDD贫血模型实体上只有 getter/setter把业务方法按行为搬到实体里服务层只做编排测试环境依赖真实数据库跑一次要三分钟没有内存仓储和测试替身用 InMemory repo fake event bus 搭建测试替身5.2 Python 的类型提示把“自由”变成“约束”动态语言落地 DDD 有一个天然劣势没有编译器帮你检查接口是否被正确实现。因此类型提示是你唯一的便宜安全保障。我在项目里会强制使用mypy或pyright做全量检查并且把仓库接口统一用Protocol定义。这样任何实现了OrderRepository协议的类如果方法签名不匹配CI 阶段直接报错。IDE 也能在你调用repo.save(order)时自动补全和提示函数签名开发效率并不输给静态语言。另外一个容易被忽略的点是Python 的dataclass在生成__init__时默认允许可变对象做字段这会让值对象的“不可变性”形同虚设。所以我在做 DDD 时坚持使用frozenTrue并且对decimal.Decimal、datetime这类不可变类型做__post_init__校验。这个习惯能避免很多隐蔽的 bug。5.3 渐进式重构建议别把老系统推倒重来最后想和你说一句真心话不要因为看了这篇文章就回去把线上系统全盘重写。DDD 落地的最佳方式是在一个边界清晰的新模块里先试点。比如老系统里有一个订单导出的功能长期被改来改去你可以先把它圈出来在domain/order/里建出聚合和用例再让接口层切换过来。等团队真的体会到“改一个业务规则只动一个地方”的爽感之后再逐步扩大范围。架构演进慢一点损失会小很多。我在实际项目里还有一个习惯每改一个需求会顺手问自己“这个改动落在哪个层领域模型需要动吗”如果答案经常是“领域模型不动只改用例”那说明边界划对了如果每次都要同时改动实体、仓储、接口那大概率是聚合边界或者上下文划分出了问题。回到业务本身去重新分析永远比埋头改代码更靠谱。