Python领域驱动设计实战：手把手实现聚合、实体与仓储

简介：一份演示领域驱动设计（DDD）落地的Java示例项目，面向中高级开发者，帮助读者理解实体、值对象、聚合、领域服务与领域事件在真实业务中的组织方式，并解决“概念会背、代码难写”的常见问题。压缩包为RAR格式，共126个文件，以Java源码、class编译产物、jar依赖库和XML配置文件为主，总大小13.72MB。其中源码对应核心聚合与领域服务实现，jar与XML用于搭建可运行环境，目录结构清晰，便于对照分析。已有1257人学习下载。通过分析账户转账、交易结算等典型业务场景的代码实现，可以掌握如何划分聚合边界、通过仓储接口隔离基础设施，以及利用领域事件实现模块解耦；还能模仿示例，将DDD设计原则迁移到订单、支付等自建业务场景。整体而言，是一份从理论到代码的桥梁式学习资料，适合希望提升业务建模与代码质量的开发者。 说实话，接触领域驱动设计（DDD）也这么多年了，我见过太多人卡在“理论看了一堆，一写代码就废”这个阶段。网上能搜到的领域模型代码示例要么是带着一堆框架注解的玩具项目，要么干脆就是披着领域模型外衣的CRUD。所以这次我打算换一种方式，不聊虚的，直接用一个完整可运行的Python领域模型示例，把实体、值对象、聚合、仓储这些概念一个一个落进代码里，顺便把我在实际项目中踩过的坑也一并交代清楚。

这个内容适合谁看？主要是两类人：一类是正在学习DDD，想看看代码层面到底怎么组织的后端开发者，另一类是已经在写业务代码，觉得Service层越来越臃肿，想找办法把业务规则从业务流程里解耦出来的朋友。我会尽量用生活化的类比把复杂概念讲明白，同时保证代码示例可以直接跑起来，你照着抄一遍，基本就能感受到领域模型和传统贫血模型的差别。

1. 领域模型到底解决什么问题

1.1 一个让我印象深刻的真实场景

我先讲一个自己经历过的案例。之前做一个订单系统，第一版图省事，所有逻辑都堆在Service里。订单有没有支付、能不能取消、库存够不够，全都靠Service里面的if-else一层层判断。刚开始订单状态少，还勉强能维护，后来业务方陆续加了“部分发货”“退款申请中”“超时自动关闭”这些状态，Service就开始失控了。一个cancelOrder方法里塞了七八个if分支，还要从不同的表里查数据，改一个逻辑要同时动好几个地方，测试用例也越来越难写。

后来我重构的时候，才真正意识到所谓领域模型，本质上就是把“这个业务规则应该由谁负责”这个问题想清楚。订单能不能取消，不应该由OrderService说了算，而应该由订单这个对象自己说了算。因为“取消订单”这件事，本身就是订单这个业务概念的一部分。这个思维的转变，才是领域模型最核心的价值。

1.2 领域模型和普通代码示例的差别

网上很多所谓的业务代码示例，本质上是“数据表驱动”的写法。实体类里只有getter和setter，没有任何行为，所有的业务逻辑都放在Service层。这种模型有个专门的名字，叫贫血模型。它的问题在于领域知识散落在各个Service中，没有和内聚的数据绑定在一起，业务复杂到一定程度就会变成一团乱麻。

而领域模型强调的是“状态和行为一体”。一个订单对象，不但有订单编号、商品列表这些属性，它还应该知道自己当前是什么状态，能执行哪些操作，在什么条件下可以转换到新状态。这与我们人类的认知方式是吻合的，就像你在现实中不会把“人”和“人会走路”拆开理解一样。后面的代码示例，我会重点展示这种差异。

2. 动手前必须先搞懂的四个概念

2.1 实体与值对象

先看这两个最容易混淆的概念。实体有唯一的标识，而且这个标识在对象的整个生命周期里都不会改变。比如一个订单，无论它的状态怎么变、商品列表怎么调整，订单编号始终不变，这个订单还是那个订单。所以实体用ID来判断相等。

值对象则没有这种标识，它完全由属性值来定义。比如订单里的金额，100元就是100元，没有人会给它单独编个号。两个金额只要币种和数值都相同，它们就是同一个东西。值对象还有个特点，它一旦创建就不可变。不是因为技术上有强迫症，而是因为值对象共享的场景很多，如果允许随意修改，就会出现一个地方改了金额、另一个地方还是旧值的诡异问题。

这里可以做一个简单对比：

2.2 聚合与仓库

聚合是领域模型里很容易被忽略、但在工程上特别重要的一环。它把一组有紧密关系的对象封装在一起，对外只暴露一个入口，这个入口叫聚合根。比如一个订单和它的订单项，订单项离开了订单就没有独立存在的意义，所以订单是聚合根，订单项是聚合内部的对象。所有对订单项的增删改查，都必须通过订单这个聚合根来操作。

这样设计的好处是，业务不变量可以在聚合内部被严格保护。比如“订单一旦确认支付，就不能再添加商品”这个规则，只要写进聚合根的方法里，不管外层是接口调用、消息监听还是定时任务，都不可能绕过这个约束。

仓储这个概念其实就是对“如何保存和获取聚合”的抽象。领域层只定义仓储接口，具体的数据库操作交给基础设施层去实现。这个模式被误解得很深，很多人以为仓储就是Repository包一层DAO，但真正的价值在于让领域层不依赖任何数据库技术，保持纯粹的业务表达。

2.3 领域服务与应用服务

有些业务操作不好归属到某个具体的实体上，比如“计算一个订单的折扣后总价”，它涉及订单本身的金额，又涉及复杂的折扣规则，放在订单对象里会让订单变得臃肿。这时候就需要领域服务。领域服务虽然是无状态的，但它在领域层内部，可以直接访问聚合和值对象。

应用服务则是另一层的东西，它负责业务流程的编排。比如“下单”这个用例，可能要调用库存服务扣减库存、调用支付服务生成支付单、再调用订单仓储保存订单。这些步骤本身不是订单对象的职责，而是业务流程。应用服务很薄，它只负责协调，不存储业务规则。很多项目把这两层混在一起，导致领域规则不断泄漏到应用层，这是特别常见的架构腐蚀信号。

2.4 模块边界怎么划

模块边界这件事，我建议遵循“一个业务用例一个聚合”的朴素原则，不要一开始就设计什么大而全的订单实体。先画一条线：哪些数据是必须一起创建、一起更新的，就放进同一个聚合；哪些只是有关联关系，但不保证同时变化的，就分开建模。

比如订单和用户，它们是两个独立的聚合，因为用户能独立存在，订单也能独立存在，它们之间通过用户ID关联，而不是通过对象引用。这样做可以让每个聚合的边界更加清晰，也方便后续拆分成独立的服务。

3. 用Python写一份可运行的领域模型示例

3.1 项目结构与依赖

我用Python中的dataclass和标准库来写这份示例，不用Django、SQLAlchemy这些重型框架，因为框架会分散注意力，让读者误认为那些ORM注解就是领域模型。项目结构如以下代码所示：

domain/ __init__.py values.py # 值对象：Money、Currency、OrderItem entities.py # 实体基类：Entity order.py # 聚合根：Order、OrderStatus domain/services/ __init__.py pricing.py # 领域服务：价格计算（满减、折扣） infra/ __init__.py repository.py # 仓储抽象接口 + 内存实现 tests/ __init__.py test_order.py # 业务规则测试

在这个结构里，domain目录不依赖任何第三方库，里面的代码只描述业务。infra目录负责基础设施，比如仓储的内存实现。tests目录验证业务规则是否符合预期。这个分层意味着，即使以后要把内存实现换成Redis或PostgreSQL，领域层代码一行都不用改。

3.2 值对象与实体的代码实现

先写值对象。金额Money是最典型的值对象，它包含了币种的校验和加法运算。我在代码里用frozen=True强制不可变，这样多个订单共享同一个金额对象也不会被意外修改。

# domain/values.py from dataclasses import dataclass from decimal import Decimal from enum import Enum class Currency(Enum): CNY = "CNY" USD = "USD" @dataclass(frozen=True) class Money: amount: Decimal currency: Currency def __post_init__(self): if self.amount < 0: raise ValueError("金额不能为负") def __add__(self, other): if not isinstance(other, Money): return NotImplemented if self.currency != other.currency: raise ValueError("币种不一致，不能相加") return Money(self.amount + other.amount, self.currency)

订单项OrderItem也是值对象，因为它没有独立标识，是订单内部的明细行。它包含商品ID、商品名称、数量和单价，并提供了一个计算小计金额的属性。

@dataclass(frozen=True) class OrderItem: product_id: str product_name: str quantity: int unit_price: Money @property def total(self) -> Money: total_amount = self.unit_price.amount * self.quantity return Money(total_amount, self.unit_price.currency)

实体基类我只保留了一个ID字段和基于ID的相等判断。订单、用户的相同性都由ID决定，而不是内存地址或属性内容。这段代码简单但必要，它让后续所有实体都具备一致的比较语义。

# domain/entities.py from dataclasses import dataclass @dataclass class Entity: id: str def __eq__(self, other): if not isinstance(other, Entity): return NotImplemented return self.id == other.id def __hash__(self): return hash(self.id)

3.3 聚合根Order的完整实现

接下来是重头戏，聚合根Order。这个类的每个方法都代表一个业务操作，而且每个操作都先检查当前状态是否允许执行。这些状态检查就是业务规则本身。我把订单状态定义为一个枚举，状态流转完全由Order对象自己控制。

# domain/order.py from dataclasses import dataclass, field from datetime import datetime from enum import Enum from .entities import Entity from .values import Money, OrderItem class OrderStatus(Enum): PENDING = "pending" PAID = "paid" SHIPPED = "shipped" CANCELLED = "cancelled" @dataclass class Order(Entity): customer_id: str items: list[OrderItem] = field(default_factory=list) status: OrderStatus = OrderStatus.PENDING paid_at: datetime | None = None created_at: datetime = field(default_factory=datetime.utcnow) def add_item(self, item: OrderItem): if self.status != OrderStatus.PENDING: raise ValueError("订单已确认，不能添加商品") if item.quantity <= 0: raise ValueError("商品数量必须大于0") for existing in self.items: if existing.product_id == item.product_id: self.items.remove(existing) item = OrderItem( product_id=existing.product_id, product_name=existing.product_name, quantity=existing.quantity + item.quantity, unit_price=existing.unit_price, ) self.items.append(item) def remove_item(self, product_id: str): if self.status != OrderStatus.PENDING: raise ValueError("订单已确认，不能移除商品") self.items = [item for item in self.items if item.product_id != product_id] def total_amount(self) -> Money: total = Money(Decimal("0"), None) for item in self.items: total = total + item.total return total def mark_paid(self): if self.status != OrderStatus.PENDING: raise ValueError("只有待支付订单可以支付") if not self.items: raise ValueError("空订单不能支付") self.status = OrderStatus.PAID self.paid_at = datetime.utcnow() def cancel(self): if self.status in (OrderStatus.SHIPPED, OrderStatus.CANCELLED): raise ValueError("当前状态下订单不能取消") self.status = OrderStatus.CANCELLED

这段代码有几个值得细看的点。

第一，total_amount方法用了Money的加法运算，而且做了一个小处理：初始值的币种先设为None，然后从第一笔明细开始累加。这样写避免了“初始化一个万能金额对象”这种尴尬设计。

第二，mark_paid和cancel都在内部检查状态，这个检查不是可有可无的形式，而是领域规则的边界。调用方可以随便尝试，但订单对象自己会拒绝非法操作。

第三，add_item在追加相同商品时会合并数量。这个逻辑如果放在Service里，谁都没把握每次调用都能想起来先检查一遍，但写在聚合根里，规则就被固定了。

3.4 仓储接口与实现

仓储接口放在领域层，它定义的是“领域需要什么样的持久化能力”，而不是“数据库怎么存”。我的订单仓储只需要三个能力：保存订单、按ID查订单、按客户查订单。

# infra/repository.py from abc import ABC, abstractmethod from domain.order import Order class OrderRepository(ABC): @abstractmethod def save(self, order: Order) -> None: ... @abstractmethod def find_by_id(self, order_id: str) -> Order | None: ... @abstractmethod def find_by_customer(self, customer_id: str) -> list[Order]: ... class InMemoryOrderRepository(OrderRepository): def __init__(self): self._store = {} def save(self, order: Order) -> None: self._store[order.id] = order def find_by_id(self, order_id: str) -> Order | None: return self._store.get(order_id) def find_by_customer(self, customer_id: str) -> list[Order]: return [order for order in self._store.values() if order.customer_id == customer_id]

这里的内存实现只为了演示，真实项目里会用SQLAlchemy、Django ORM或者MongoDB客户端来替换它。替换的关键是，Order对象本身不含任何数据库字段，比如不会有db_created_at或者_sa_instance_state这类东西，所以领域层是干净的。

3.5 用测试验证业务规则

最后一个环节，用单元测试把关键业务规则固化下来。我推荐把测试当作领域模型的说明书，每个测试方法都对应一条业务规则。

# tests/test_order.py from datetime import datetime from decimal import Decimal import pytest from domain.order import Order, OrderStatus from domain.values import Currency, Money, OrderItem def build_order(order_id="order-001", customer_id="customer-001"): order = Order(id=order_id, customer_id=customer_id) item = OrderItem( product_id="p-001", product_name="测试商品", quantity=2, unit_price=Money(Decimal("50.00"), Currency.CNY), ) order.add_item(item) return order def test_add_item_with_duplicate_product_merges_quantity(): order = build_order() new_item = OrderItem( product_id="p-001", product_name="测试商品", quantity=3, unit_price=Money(Decimal("50.00"), Currency.CNY), ) order.add_item(new_item) assert len(order.items) == 1 assert order.items[0].quantity == 5 def test_order_cannot_add_item_after_paid(): order = build_order() order.mark_paid() with pytest.raises(ValueError): order.add_item( OrderItem( product_id="p-002", product_name="新商品", quantity=1, unit_price=Money(Decimal("10.00"), Currency.CNY), ) ) def test_order_cancel_after_shipped_is_rejected(): order = build_order() order.mark_paid() order.status = OrderStatus.SHIPPED with pytest.raises(ValueError): order.cancel()

运行这些测试，会发现它们不依赖数据库、不依赖网络，秒级执行完。这其实是领域模型带来的一个隐性收益，你可以在不启动任何服务的情况下，快速验证业务规则的正确性，CI里跑测试也特别快。

4. 代码示例之外的经验与思考

4.1 常见问题速查表

我在带团队做DDD落地时，发现大家问得最多的问题基本集中在下面这几类：

这里我想特别强调一下第二个问题。如果你在项目里直接用Django的Model或者SQLAlchemy的Declarative类当领域实体，那其实是把基础设施的束缚带进了领域层。领域模型应该用纯Python对象，ORM模型在仓储内部做转换，两者分离，这个代价换取的是领域层的长期稳定。

注意：千万不要为了追求“纯正的DDD”而在项目里强行套一堆模式。一个导出Excel报表的功能、一个简单的字典查询接口，完全不需要聚合和仓储。领域模型模式适合业务规则复杂的核心域，工具型代码用过程式写法更合适。

4.2 领域模型思想在其他技术栈的迁移

这份Python示例只是载体，领域模型的思想完全可以平移到其他技术栈。比如在Android的开发中，MVVM架构里的ViewModel层最适合承载应用服务编排，而Model层则可以用领域对象来表达业务规则。我见过一些项目在MVVM的Model层使用Java的POJO，里面没有任何行为，所有判断写在ViewModel里，结果就是ViewModel几千行，测试只能靠手机手工点。

换到.NET生态的WPF或ASP.NET Core，也可以用相同的思路，实体保持行为、仓储抽象、应用服务编排。再说一个偏工程的方向，如果你写过用脚本处理Excel文件的小工具，比如读取上百个工作表然后汇总统计，这个场景适合用“值对象+纯函数”的方式组织逻辑，把“一行数据解析结果”建模成不可变的值对象，再丢给统计函数处理，测试起来比传统的foreach里改一堆全局变量要舒服得多。

从语言层面看，Python的bool、int、str本身就是不可变的值对象，天天都在用。只是很多人在建模业务时把值对象这个工具忘了，一切都写成可变的字典或实体类。

4.3 我踩过的坑和建模心得

最后分享几个真金白银换回来的教训。

第一个坑是“从数据库表反向设计聚合”。我刚开始做DDD的时候，习惯先从数据库设计入手，先建表、再根据表写实体，结果设计出来的聚合其实还是围绕数据表来的，完全违背了领域建模的初衷。正确的方式是先从业务规则出发，找出不变量，再设计聚合和边界，最后才考虑表结构。

第二个坑是“过度建模”。有一次做一个营销活动模块，我设计了一堆领域事件、规格模式、策略模式，结果业务变化并不频繁，代码反而因为抽象层级太多变得难以维护。后来我总结出一个原则：先给一个流畅的聚合内部实现，等业务确实出现第二处需要相同规则的地方，再抽取抽象。过早抽象是万恶之源。

第三个心得是，领域模型的边界意识要强。状态机就是很好的例子，如果你把订单状态的判断分散在各个业务用例里，那么每次新增状态都要像大海捞针一样找所有相关代码。而把状态转换收拢到聚合根内部之后，新增状态只需要在Order类里加一个方法或一个枚举值，影响范围就固定了。

最后提一句测试策略。领域层的测试应该像这份示例里的测试一样，只针对业务规则，不涉及框架、数据库、网络。测的不是“接口是否返回200”，而是“订单已支付后能不能再添加商品”。实践下来，这类测试的稳定性和可维护性都远高于Controller层的集成测试。你把领域规则测稳了，上层应用服务的编排即使重构也不容易出大问题。

本文还有配套的精品资源，点击获取