在python web开发中践行ddd需五步:一、领域模型与框架解耦;二、应用服务协调用例;三、仓储接口抽象持久化;四、领域事件驱动最终一致性;五、防腐层适配外部系统。

如果您希望在 Python Web 框架中引入领域驱动设计(DDD)思想,但发现框架默认结构与 DDD 分层理念存在张力,则可能是由于传统 MVC 或路由驱动模式掩盖了领域模型的边界与职责。以下是将 DDD 核心要素融入 Python Web 开发的具体实践路径:
一、分离领域模型与框架胶水代码
该方法旨在确保领域实体、值对象、聚合根等核心模型完全脱离 Web 框架的生命周期管理,避免 Flask 或 FastAPI 的请求上下文、装饰器、依赖注入机制污染领域逻辑。模型应可独立单元测试,不导入任何框架模块。
1、在项目根目录下创建 domain/ 包,内含 models/、exceptions/、services/ 子包,所有文件禁止 import flask、fastapi、starlette 等框架符号。
2、定义聚合根类时,使用纯 Python 属性与私有方法封装业务不变量,例如 Order 类中通过 _items 列表与 _validate_total() 方法强制校验订单总额一致性。
立即学习“Python免费学习笔记(深入)”;
3、Web 层(如 FastAPI 的 router 模块)仅通过构造函数或工厂函数接收已实例化的领域对象,绝不调用 request.state 或 Depends() 注入领域类本身。
二、实现应用服务层作为用例协调器
该方法将用户场景(如“创建订单”、“取消订阅”)封装为无状态的应用服务,其唯一职责是编排领域对象、调用领域服务、触发领域事件,并将结果转译为框架可理解的响应格式。它不包含业务规则,仅作流程胶合。
1、在 application/ 包中定义 order_service.py,其中 create_order() 函数接收原始参数(如 user_id、item_list),不接收 request 对象。
2、函数内部调用 Order.create() 构建聚合,再调用 inventory_service.reserve()(领域服务接口)检查库存,失败则抛出 InsufficientStockError。
3、成功后调用 event_bus.publish(OrderCreatedEvent(order_id)),事件对象为数据类,不含框架相关字段。
三、采用仓储接口与具体实现解耦
该方法通过抽象仓储协议(Protocol 或 ABC)隔离领域层对持久化技术的依赖,使领域代码无需知晓 SQL、MongoDB 或内存存储细节,便于替换基础设施并支持测试替身。
1、在 domain/repositories.py 中定义 OrderRepository 协议,声明 add()、by_id()、find_by_status() 等方法签名,返回类型标注为 Order 或 List[Order]。
2、在 infrastructure/repositories.py 中实现 SQLAlchemyOrderRepository,继承 OrderRepository 并注入 session,内部使用 ORM 映射,但对外不暴露 Session 或 Model 类型。
3、应用服务通过依赖注入获取 OrderRepository 实例,运行时由容器绑定具体实现,领域层代码保持零引用 infrastructure 包。
四、引入领域事件与最终一致性处理
该方法利用领域事件显式表达业务事实的发生,将跨聚合的副作用(如发送通知、更新统计)移出主事务,交由异步处理器执行,保障核心流程的确定性与性能。
1、在 domain/events.py 中定义不可变事件类,如 OrderPaidEvent,包含 order_id: UUID、paid_at: datetime 字段,无方法体。
2、聚合根在状态变更点(如 pay() 方法内)调用 self._record_event(OrderPaidEvent(...)),事件暂存于聚合内部列表,不立即分发。
3、应用服务在事务提交前,遍历所有聚合的事件列表,调用 event_bus.publish_all(aggregate.pop_events()),由总线将事件转发至注册的处理器,如 send_payment_confirmation_email。
五、构建防腐层适配外部系统契约
该方法针对第三方 API、遗留系统或不符合领域语义的内部服务,建立翻译转换层,确保外部数据结构、错误码、分页机制等不侵入领域模型,维持领域语言的纯粹性。
1、在 adapters/external/ 下创建 payment_gateway.py,定义 PaymentGatewayClient 类,封装 HTTP 调用与 JSON 解析逻辑。
2、提供 charge_card() 方法,输入为领域对象 PaymentIntent,输出为 PaymentResult 数据类;内部将 PaymentIntent.amount_cents 映射为 "amount" 字段,将网关返回的 "status": "succeeded" 转译为 PaymentResult.success = True。
3、领域服务(如 payment_service.py)仅依赖 PaymentGatewayClient 接口,具体实现由基础设施层提供,领域代码不接触 requests.Session 或 response.json()。











