做后端时间一长很多人都会遇到同一个困扰代码量不大时一切都很清爽一旦业务复杂起来model层越来越厚service层变得又臭又长一个函数几十个if改一个需求像拆炸弹。我也经历过这个阶段。搜索“ddd架构”、“ddd搞不懂”的人越来越多说明大家对这个词有期待但又被它的各种概念劝退了。这篇文章想聊的是在 Python 项目里DDD领域驱动设计到底怎么落地而不陷入“为了架构而架构”的泥潭。我不会跟你复述教科书上的定义也不打算把整套 Eric Evans 的理论搬一遍。我会从一个真实的电商订单场景出发给你一套可以直接抄的 Python 代码骨架告诉你哪些层该拆、哪些代码该放哪、事务和 ORM 怎么配合顺便把最容易踩的坑也列出来。适合这些人看写过 Flask/Django/FastAPI 业务代码、感觉逻辑越来越难维护、听说过 DDD 但不知道怎么动手的开发者。1. 先弄明白DDD 到底在解决什么问题1.1 一个真实场景订单系统的失控之路我见过很多 Python 项目都是这么“长歪”的。一开始你接手的是一个小订单系统用户下单、付款、发货三张表。此时你写的是经典的Model Service结构比如用 Djangodef create_order(request): cart request.user.cart order Order.objects.create( userrequest.user, total_amountcart.total(), statuspending ) 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有状态会经历“待付款、已支付、已发货、已完成、已取消”这些变化。而我们通常说的“金额”则是一个值对象由数值和币种组成两个金额的比较不是比地址而是比数值。先看值对象我用dataclassfrozenTrue来实现不可变性from dataclasses import dataclass from decimal import Decimal from typing import Union dataclass(frozenTrue) 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_factorylambda: str(uuid4())) items: list[OrderItem] field(default_factorylist) status: OrderStatus OrderStatus.PENDING created_at: datetime field(default_factorydatetime.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(frozenTrue) class DomainEvent: event_id: str field(default_factorylambda: str(uuid4())) occurred_at: datetime field(default_factorydatetime.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_idself.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_iduser_id) for item in items: product self._product_repo.get(item.sku_id) order.add_item( product_idproduct.sku_id, priceproduct.price, quantityitem.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_idorder_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_idrow.order_id, items[ OrderItem(product_idi.product_id, priceMoney(Decimal(i.amount), i.currency), quantityi.quantity) for i in row.items ], statusOrderStatus(row.status), created_atrow.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)}, headersself._headers(), timeout5, ) # 把外部返回包装成 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 懒加载与 N1 查询领域层拿到Order之后如果OrderRow.items是懒加载很可能在遍历订单项时触发 N1 查询。这种问题在“展示层读数据”时特别明显。建议查询场景不要复用仓储接口单独写查询模型或 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_repoSqlAlchemyOrderRepository(session_factory), product_repoSqlAlchemyProductRepository(session_factory), event_busevent_bus, )然后再app.post(/orders) def place_order(input: CreateOrderRequest, use_case: PlaceOrderUseCase Depends(get_place_order_use_case)): order_id use_case.execute(user_idinput.user_id, itemsinput.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 带来的最大收益并不是“代码变得多优雅”而是需求变更时心里有底了你知道该改哪个文件不会影响别人你知道新增一种订单状态时只需要动状态机那一张表。这种确定性是面对长期维护的项目最值钱的东西。