做后端时间一长，很多人都会遇到同一个困扰：代码量不大时一切都很清爽，一旦业务复杂起来，model层越来越厚，service层变得又臭又长，一个函数几十个if，改一个需求像拆炸弹。我也经历过这个阶段。搜索“ddd架构”、“ddd搞不懂”的人越来越多，说明大家对这个词有期待，但又被它的各种概念劝退了。
这篇文章想聊的是：在 Python 项目里，DDD（领域驱动设计）到底怎么落地，而不陷入“为了架构而架构”的泥潭。
我不会跟你复述教科书上的定义，也不打算把整套 Eric Evans 的理论搬一遍。我会从一个真实的电商订单场景出发，给你一套可以直接抄的 Python 代码骨架，告诉你哪些层该拆、哪些代码该放哪、事务和 ORM 怎么配合，顺便把最容易踩的坑也列出来。适合这些人看：写过 Flask/Django/FastAPI 业务代码、感觉逻辑越来越难维护、听说过 DDD 但不知道怎么动手的开发者。
1. 先弄明白：DDD 到底在解决什么问题
1.1 一个真实场景：订单系统的失控之路
我见过很多 Python 项目都是这么“长歪”的。
一开始，你接手的是一个小订单系统：用户下单、付款、发货，三张表。此时你写的是经典的Model + Service结构，比如用 Django：
def create_order(request): cart = request.user.cart order = Order.objects.create( user=request.user, total_amount=cart.total(), status='pending' ) return order这没什么问题。但后来需求开始膨胀：优惠券、会员折扣、库存预占、退款审核、第三方支付回调、超时自动取消。慢慢地，你的OrderService变成了一个上千行的上帝类，每个方法里十几个if嵌套，改一个规则要翻半天的代码，测试也不好写，线上还时不时冒出一个“订单状态不对”的 bug。
这段失控的根源，不是代码写得不够好，而是业务逻辑散落在了各处。状态判断在 service 里、金额计算散在 util 里、内部校验散在 Controller 里。你缺少一个把业务规则“收拢”的地方，也缺少一套描述业务边界的语言和结构。
DDD 的用途就是：把业务领域里的核心逻辑从技术细节中剥离出来，放进一个独立的领域层，让它能独立演进、独立测试。听着很玄，核心做法其实就两件事：找到边界、划清职责。
1.2 把 DDD 的专业术语翻译成人话
很多人在学 DDD 时，是被一堆术语劝退的。这里我用白话来翻译一遍，先用着，后面代码里你会看到它们长什么样。
- 领域（Domain）：你要解决的业务问题的范围。订单、支付、库存，都是领域的一部分。
- 实体（Entity）：有唯一标识（ID）、会改变状态的对象。一个订单就是实体，因为订单
OrderId固定，但状态会变。 - 值对象（Value Object）：没有唯一标识，靠属性值本身来定义和比较的对象。金额
Money、地址Address都是典型的值对象，两个“100元”不需要区分是哪个 100 元。 - 聚合（Aggregate）：一组必须保持一致性的对象集合。比如订单和订单项，删除订单时必须同时处理订单项，这种“一致性边界”就是聚合。
- 聚合根（Aggregate Root）：聚合中对外的入口对象。操作订单项不能绕过订单，订单就是聚合根。
- 仓储（Repository）：获取和存储聚合的接口，它让你在业务代码里不需要直接操作数据库。
- 应用服务（Application Service）：编排用例的“导演”，负责拿数据、调领域模型、保存结果、提交事务。
- 领域服务（Domain Service）：不适合放在某个实体或值对象上的领域逻辑，比如“计算整个订单的优惠分摊”。
- 防腐层（Anti-Corruption Layer）：隔离外部系统（第三方 API、遗留系统）与你的领域模型的翻译层。
记住这张心智地图：应用服务调度领域层，领域层产出结果，仓储接口去持久化，基础设施层提供实现。依赖方向永远是“业务向内，技术向外”。
2. 搭建项目的四层骨架：目录结构 + 依赖方向
2.1 目录结构到底怎么放
我在 FastAPI 项目里常用的结构是这样：
order_service/ ├── interfaces/ # 接口层：HTTP路由、请求/响应模型、中间件 │ └── http/ │ ├── api/ │ ├── models.py │ └── main.py ├── application/ # 应用层：用例编排、事务边界 │ ├── services/ │ └── dtos/ ├── domain/ # 领域层：核心业务逻辑 │ ├── models/ # 实体、值对象、聚合根 │ ├── services/ # 领域服务 │ ├── events/ # 领域事件 │ └── repositories/ # 仓储接口（抽象） ├── infrastructure/ # 基础设施层：数据库、外部API、消息等 │ ├── repositories/ # 仓储的 SQLAlchemy / Django ORM 实现 │ ├── cache/ │ └── externals/ # 第三方客户端 └── common/ # 公共类型、异常、工具这个结构不是死板的标准，而是对“按层组织”和“按模块组织”的折中。业务足够大时，比如你同时做订单、库存、用户三个模块，建议升级为“按子领域分包”：
order_service/ ├── interfaces/http/ ├── application/ ├── domain/ └── infrastructure/每个子领域内部再按四层去铺。这样模块之间天然多了边界，不太会写串。
2.2 为什么依赖方向必须是“向内的”
四层结构最容易犯的错误是：领域层里的模型直接用了 SQLAlchemy 的session、Django 的Model.objects，导致业务代码和 ORM 绑死。DDD 强调的依赖倒置在这时候起作用：领域层只定义仓储接口，然后由基础设施层去实现它。业务代码面向接口编程，数据库怎么换都不会动到领域逻辑。
举个例子，我在domain/repositories/order_repository.py里定义：
from abc import ABC, abstractmethod class OrderRepository(ABC): @abstractmethod def get(self, order_id: str) -> "Order": ... @abstractmethod def save(self, order: "Order") -> None: ...然后在infrastructure/repositories/order_repository.py里实现：
class SqlAlchemyOrderRepository(OrderRepository): def __init__(self, session_factory): self._session_factory = session_factory def get(self, order_id: str) -> Order: with self._session_factory() as session: # 把 ORM 模型映射成领域模型 ... def save(self, order: Order) -> None: with self._session_factory() as session: ...这样领域层完全不知道数据库是什么，测试时可以随便用一个InMemoryOrderRepository来替代。
依赖关系用一句话总结：外层依赖内层，内层依赖抽象，而不是具体实现。
3. 领域层实战：把业务逻辑真正写进模型
3.1 实体与值对象的代码示例
拿电商订单来说。“订单”本身是一个实体。它有 ID，有状态，会经历“待付款、已支付、已发货、已完成、已取消”这些变化。而我们通常说的“金额”则是一个值对象，由数值和币种组成，两个金额的比较不是比地址，而是比数值。
先看值对象，我用dataclass+frozen=True来实现不可变性：
from dataclasses import dataclass from decimal import Decimal from typing import Union @dataclass(frozen=True) class Money: amount: Decimal currency: str = "CNY" def __add__(self, other: "Money") -> "Money": if self.currency != other.currency: raise ValueError("币种不一致，不能相加") return Money(self.amount + other.amount, self.currency) def __mul__(self, times: Union[int, Decimal]) -> "Money": return Money(self.amount * Decimal(str(times)), self.currency) def is_zero(self) -> bool: return self.amount == 0金额计算如果不封装成值对象，散落在各个 service 里会特别容易出现精度问题和币种混淆。用Money统一之后，加法、乘法都走同一个逻辑。
再看实体。我习惯把实体设计成“有行为”的类，而不是单纯承载数据：
from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import Enum from uuid import uuid4 class OrderStatus(str, Enum): PENDING = "pending" PAID = "paid" SHIPPED = "shipped" COMPLETED = "completed" CANCELLED = "cancelled" @dataclass class OrderItem: product_id: str price: Money quantity: int @dataclass class Order: order_id: str = field(default_factory=lambda: str(uuid4())) items: list[OrderItem] = field(default_factory=list) status: OrderStatus = OrderStatus.PENDING created_at: datetime = field(default_factory=datetime.utcnow) def add_item(self, product_id: str, price: Money, quantity: int) -> None: if self.status != OrderStatus.PENDING: raise DomainRuleViolation("只有待支付订单才能修改商品") self.items.append(OrderItem(product_id, price, quantity)) def paid(self) -> None: if self.status != OrderStatus.PENDING: raise DomainRuleViolation("只有待支付订单才能付款") if self.total().amount <= 0: raise DomainRuleViolation("空订单不能支付") self.status = OrderStatus.PAID def cancel(self) -> None: if self.status in (OrderStatus.SHIPPED, OrderStatus.COMPLETED): raise DomainRuleViolation("已发货或已完成订单不能取消") self.status = OrderStatus.CANCELLED def total(self) -> Money: total = Money(Decimal("0")) for item in self.items: total = total + item.price * item.quantity return total这段代码包含了两层含义：一是你能看出Order对外暴露的行为是什么，二是业务规则被限制在模型内部，非法状态转移会直接抛异常。这样写的好处是，到时候找“订单不能取消”的规则，不用满工程搜if status == ...，看模型方法就行。
3.2 聚合与一致性边界：先画清楚再写代码
设计聚合是 DDD 落地里最需要经验的部分。一个简单的方法是：问自己，哪组对象必须同时变更？
拿订单来说，订单和订单项显然是一个聚合，订单项不能脱离订单独立存在，删除订单时订单项也必须删除。所以OrderItem是聚合内部实体，Order是聚合根。对外不要暴露直接修改items列表的方法，而是通过add_item、remove_item这样的聚合根方法去操作。
不少初学者在这个地方会走偏，觉得聚合根就是把所有数据放在一个类里。于是什么字段都往Order上塞，最后Order又变成一个大杂烩。正确的思路是：聚合边界不只是“包含”，更重要的是一致性约束的边界。
比如“库存预占”，它在业务上跟订单项有紧密关系，但库存预占和订单本身并不一定属于同一个聚合，因为它们的生命周期不一样，锁粒度也不一样。如果把库存也塞进订单聚合里，你会在大量并发场景下死锁。所以我的经验是：当两个对象一致性要求高、变化频繁一起发生时，才放进同一个聚合；否则就拆分聚合，用领域事件来异步协调。
一个聚合里只保留这一个边界内必须一致的东西，代码会清爽很多。
3.3 领域服务与领域事件：处理跨聚合的复杂逻辑
有些业务动作没法合理地放在某个实体上，比如“计算优惠券分摊到每个订单项上”。它是领域逻辑，但需要同时读写多个聚合甚至多个值对象。这时候就写一个领域服务。
class PromotionCalculationService: def apply_promotion(self, order: Order, promotion_code: str) -> Money: # 复杂规则：按比例折扣、阶梯优惠、满减 … # 返回优惠金额 ...领域服务本身不持有状态，它只是承载“动作”。判断标准很简单：如果这段逻辑强行塞进实体里会让实体违背单一职责，或者逻辑不只依赖一个对象，那就抽出来。
领域事件是我在这个阶段比较推荐引入的机制。比如“订单已支付”之后，需要发短信、扣库存、开电子发票。如果这些都用同步调用的方式挂在Order.paid()方法里，等于把领域模型和外部副作用耦合在一起。更合理的方式是：领域模型只负责产出事件，由外部订阅方去处理。事件总线的实现不一定要引入消息队列，先在应用内做一个简单的同步总线就够用：
from dataclasses import dataclass from typing import Callable, dict @dataclass(frozen=True) class DomainEvent: event_id: str = field(default_factory=lambda: str(uuid4())) occurred_at: datetime = field(default_factory=datetime.utcnow) class DomainEventBus: def __init__(self): 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: DomainEvent) -> None: for handler in self._subscribers.get(type(event), []): handler(event)在Order.paid()里发布事件：
@dataclass class OrderPaidEvent(DomainEvent): order_id: str # 在 Order.paid() 中 self.status = OrderStatus.PAID self.events.append(OrderPaidEvent(order_id=self.order_id))然后用一个工作单元把这些事件统一发出去。这块做到后面，你会发现订单状态机和“支付后要做的那些事”彻底解耦了。
4. 应用层与基础设施层：把用例和技术细节接起来
4.1 应用服务的正确打开方式
应用服务不是“业务逻辑的家”，它是用例的编排器。它做四件事：
- 从仓储拿领域模型；
- 调用领域模型的方法完成业务动作；
- 通过事件总线发出领域事件；
- 保存结果，通常是提交事务。
来看一个“下单”用例的伪代码：
class PlaceOrderUseCase: def __init__(self, order_repo: OrderRepository, product_repo: ProductRepository, event_bus: DomainEventBus): self._order_repo = order_repo self._product_repo = product_repo self._event_bus = event_bus def execute(self, user_id: str, items: list[OrderItemInput]) -> str: order = Order(user_id=user_id) for item in items: product = self._product_repo.get(item.sku_id) order.add_item( product_id=product.sku_id, price=product.price, quantity=item.quantity, ) self._order_repo.save(order) for event in order.events: self._event_bus.publish(event) return order.order_id体会一下：这里没有if order.status == ...，没有try: db.session.commit()，也没有request对象。用例类清晰描述了业务的“剧情”，而具体怎么存、怎么通知的细节都被隔离到了外部。
4.2 仓储接口与 SQLAlchemy 的实现细节
在 Python 生态里，SQLAlchemy和Django ORM是两种主流方案，但它们与 DDD 配合时都有各自的摩擦点。最大的摩擦在于：ORM 模型通常都有「隐式的会话连接」和「延迟加载（lazy loading）」，而领域模型是不希望感知这些的。
我的做法是：ORM 模型只作为持久化映射，领域模型是独立的普通类，两者在仓储层做转换。
这是infrastructure/repositories/order_repository.py里的一段典型实现：
class SqlAlchemyOrderRepository(OrderRepository): def __init__(self, session_factory): self._session_factory = session_factory def get(self, order_id: str) -> Order: with self._session_factory() as session: row = session.query(OrderRow).filter_by(order_id=order_id).first() if not row: raise OrderNotFound(order_id) return self._to_domain(row) def save(self, order: Order) -> None: row = self._to_row(order) with self._session_factory() as session: session.merge(row) session.commit() def _to_domain(self, row: OrderRow) -> Order: return Order( order_id=row.order_id, items=[ OrderItem(product_id=i.product_id, price=Money(Decimal(i.amount), i.currency), quantity=i.quantity) for i in row.items ], status=OrderStatus(row.status), created_at=row.created_at, )需要注意几个点：
_to_domain和_to_row是转换器，这种手写代码看起来有点啰嗦，但作用是清晰的：领域模型不污染 ORM 映射；session.merge而不是session.add，是为了处理可能存在的既有记录，避免主键冲突；- 仓储层方法粒度尽量与聚合一致，比如
save(order)直接保存整个聚合，而不是save_order_item(item)这样的细粒度方法。
测试时，你可以写一个内存仓储放在测试模块里，速度飞快，还不用搭数据库，这是面向接口设计的红利。
4.3 用状态机代替四处开花的 if-else
订单状态流转是 DDD 落地中很典型的一个“价值点”。如果不用状态机，通常你会写出这样的代码：
if order.status == OrderStatus.PENDING and action == "pay": ... elif order.status == OrderStatus.PAID and action == "ship": ...当状态越来越多时，这些if会蔓延到接口层、应用层、甚至前端配合改。更稳的写法是引入一个简单的状态机：
class OrderStateMachine: transitions = { OrderStatus.PENDING: {OrderEvent.PAY: OrderStatus.PAID, OrderEvent.CANCEL: OrderStatus.CANCELLED}, OrderStatus.PAID: {OrderEvent.SHIP: OrderStatus.SHIPPED}, OrderStatus.SHIPPED: {OrderEvent.COMPLETE: OrderStatus.COMPLETED}, } @classmethod def next(cls, current: OrderStatus, event: OrderEvent) -> OrderStatus: try: return cls.transitions[current][event] except KeyError: raise DomainRuleViolation(f"不能从状态 {current} 触发事件 {event}")然后把实体里的状态变更方法改成调用这个状态机，比如：
def paid(self): self.status = OrderStateMachine.next(self.status, OrderEvent.PAY)这样以后要加新的状态机规则，只需要改transitions字典，业务代码的阅读成本会大幅降低。这是我踩过很多坑之后才学会的写法——当时线上有人直接把已取消的订单改成了已支付，就是状态漫无止境的if判断漏了一处。
4.4 防腐层：和外部系统划清界限
在实际业务中，订单系统一定会对接外面的系统：支付网关、ERP、物流平台。这些外部系统如果直接被领域层引用，领域逻辑就会被外部 API 的细节绑架。
防腐层的思路是：在领域层定义一套“内部语言”的接口，在基础设施层实现适配器，把外部系统的模型翻译成领域模型。
例如，支付领域里我们定义：
class PaymentGateway(ABC): @abstractmethod def create_payment(self, order_id: str, amount: Money) -> PaymentResult: ... @abstractmethod def query_payment(self, payment_id: str) -> PaymentStatus: ...然后在infrastructure/externals/union_pay_gateway.py里实现真正的 HTTP 调用：
class UnionPayGateway(PaymentGateway): def create_payment(self, order_id: str, amount: Money) -> PaymentResult: resp = requests.post( self._api_url, json={"order_id": order_id, "amount": str(amount.amount)}, headers=self._headers(), timeout=5, ) # 把外部返回包装成 PaymentResult ...这样，即使支付渠道换了，领域层的业务流程不受影响，只需要换掉PaymentGateway的实现即可。
5. Python 项目落地 DDD 的典型问题与排查记录
5.1 别把 DDD 做成全项目“一刀切”
这是我最大的心得：不是所有模块都需要用 DDD 来约束。像纯 CRUD 的字典表、白名单、简单的配置读取，用传统 MVC 方式反而更高效。过度设计是 Python 社区里接触到 DDD 后最容易犯的毛病——一个简单的博客却搞了聚合、事件总线、CQRS，最后自己都被绕晕了。
我在项目里的取舍标准是：
- 业务复杂度高（订单、支付、库存、审批流）→ 上领域层；
- 规则简单（字典表、如用户简单资料维护）→ 直接走 FastAPI + SQLAlchemy 就是最快的；
- 状态很多、规则经常变→ 领域层 + 状态机；
- 报表/查询场景→ 不要走仓储和聚合，直接用 QueryService 去查库或查缓存，这就是为什么很多 DDD 项目同时会引入 CQRS 的查询侧。
一上来不建议做全系统重构，更稳妥的方式是选一条业务线（比如订单）先做改造试点，跑通后再复制模式。
5.2 事务边界不知道放哪里
DDD 里很容易犯的一个错误是：把事务控制放到仓储或领域模型方法里。比如让Order.paid()在内部去调用db.session.commit()，这会让领域模型与数据库耦合，也会在聚合根之间产生方向不明的事务依赖。
我更推荐的做法是：事务边界放在应用服务层。因为一个用例通常对应一个事务，在处理完领域操作后统一提交。遇到跨聚合的事务，尽量通过最终一致性和领域事件来解耦，不要直接用一个大事务锁住所有表。
如果在 FastAPI 里，我常用Depends的方式在应用服务外包装事务，把session_factory注入到用例里，让用例自己决定 commit 或 rollback。
5.3 ORM 懒加载与 N+1 查询
领域层拿到Order之后，如果OrderRow.items是懒加载，很可能在遍历订单项时触发 N+1 查询。这种问题在“展示层读数据”时特别明显。
建议：查询场景不要复用仓储接口，单独写查询模型或 DTO。需要显示订单列表及商品、总金额时，直接写一个 SQLAlchemy join 查询，只查询必要的列并映射成 DTO，而不是先查出 N 个订单再逐一懒加载订单项。这就是 CQRS 的“读模型分离”。只要能改善性能，往读侧倾斜一点是值得的。
几年前我做一个订单管理后台时，列表加载慢到 3 秒，原因就是遍历了 200 个聚合根又触发懒加载。改成按需查询 DTO 后，接口响应降到 200 毫秒以内，这个优化立竿见影。
5.4 动辄就满屏障的抽象工厂和依赖注入
Python 不是 Java，不需要把所有依赖都交给容器。DDD 落地时，简单的“手动组装”反而更容易让人看懂。尤其是 FastAPI 的Depends已经提供了不错的依赖管理能力，你完全可以在路由层这样组装：
def get_place_order_use_case(): return PlaceOrderUseCase( order_repo=SqlAlchemyOrderRepository(session_factory), product_repo=SqlAlchemyProductRepository(session_factory), event_bus=event_bus, )然后再：
@app.post("/orders") def place_order(input: CreateOrderRequest, use_case: PlaceOrderUseCase = Depends(get_place_order_use_case)): order_id = use_case.execute(user_id=input.user_id, items=input.items) return {"order_id": order_id}没必要引入一整套 dependency injection framework，除非团队对它已经非常熟悉。强行加容器只会让排查问题的时候多一层跳转。
5.5 统一语言要“活在代码里”，而不是停留在文档里
DDD 里强调“统一语言（Ubiquitous Language）”，很多团队的做法是开会讨论了半天，结果代码里的类名、字段名还是当初数据库表的名字。
我的建议是，术语要落到模型名、方法名、事件名和状态名上。比如业务里说的是“待付款”，代码里就用PENDING_PAYMENT而不是STATUS_1；业务说“取消订单”，代码里就用cancel()而不是change_status(3)。命名对齐后，业务人员和开发开会时效率会高很多，因为你说的每一个词都能在代码里找到对应的东西。这是比架构更便宜、也更持久的收益。
另一个经常踩的坑是：领域事件直接用消息队列的 topic 名来命名，比如order.pay.success，这会让领域层暴露基础设施细节。我习惯在领域层发布时用纯业务名，如OrderPaidEvent，然后在基础设施层的监听器里再映射到具体的 MQ topic。
写在最后的体会
如果你现在正处在一个 Python 业务代码开始“失控”的阶段，我的建议是从最小的一步开始：为你的核心模块划出domain目录，把那些反复出现的状态判断和金额计算先搬进模型；再挑一条业务线，用“仓储 + 应用服务”的方式重写出一个用例；最后再考虑事件、状态机这些进阶武器。每做一步，都先问自己一个问题——这段核心业务逻辑，离开 Web 框架和数据库之后，还能不能独立测试？如果能，你的 DDD 基本已经落地了。
在我自己经历过的项目里，DDD 带来的最大收益并不是“代码变得多优雅”，而是需求变更时心里有底了：你知道该改哪个文件不会影响别人，你知道新增一种订单状态时只需要动状态机那一张表。这种确定性，是面对长期维护的项目最值钱的东西。
