摘要
本文深入探讨领域驱动设计在 Python 中的完整实现方案，聚焦实体、值对象、聚合、仓储等核心概念。通过架构流程图、完整可运行代码示例和企业级实战案例，展示如何利用 Python 生态工具构建高可维护的复杂业务系统。文章包含性能优化技巧、常见问题解决方案以及 Pydantic 在 DDD 中的高级应用，为 Python 开发者提供从入门到精通的完整指南。
1 引言：为什么 Python 开发者需要领域驱动设计
在我多年的 Python 开发生涯中，见证了太多项目从快速原型演变为难以维护的'大泥球'。记得曾经接手一个金融交易系统，业务逻辑散落在 Django 视图、SQLAlchemy 模型和工具函数中，简单的需求变更需要修改 10 多个文件。引入 DDD 后，代码复杂度降低了 60%，新功能开发速度提升了 3 倍，这让我深刻认识到 DDD 在复杂业务系统中的价值。
1.1 Python 与 DDD 的天然契合度
很多人认为 Python 作为动态语言不适合大型项目架构，这是严重的误解。Python 的灵活性恰恰是实施 DDD 的优势：
# Python 的动态特性让领域模型表达更自然
from dataclasses import dataclass
from typing import List
@dataclass
class Order:
id: str
items: List['OrderItem']
status: str
def calculate_total(self) -> float:
return sum(item.price * item.quantity for item in self.items)
# 业务规则直接体现在领域模型中
def can_be_cancelled(self) -> bool:
return self.status in ['created', 'pending']
与 Java 等静态语言相比，Python 的鸭子类型、注解支持和丰富的语法糖让领域模型的实现更加简洁直观。
1.2 DDD 在 Python 生态中的成熟度
近年来，Python 生态对 DDD 的支持日益完善：
- FastAPI：原生支持依赖注入和 Pydantic 模型
- SQLAlchemy 2.0：改进的 ORM 模式与领域模型更好兼容
- Pydantic v2：强大的数据验证和序列化能力
- 各类 DDD 框架：如 dddpy、python-ddd 等专门支持
下面通过架构演进图展示 DDD 如何解决传统 Python 项目的痛点：
2 DDD 核心概念与 Python 实现
2.1 实体：具有生命周期的业务对象
实体是 DDD 中最基础的概念，它通过唯一标识来区分对象，而不是通过属性值。
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from datetime import datetime
@dataclass
class Product:
"""产品实体 - 具有唯一标识和生命周期"""
id: UUID = field(default_factory=uuid4)
name: str
description: str
price: float
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
def change_price(self, new_price: float) -> None:
"""改变价格 - 实体行为"""
if new_price <= 0:
raise ValueError("价格必须大于 0")
self.price = new_price
self.updated_at = datetime.now()
def update_description(self, description: str) -> None:
"""更新描述 - 实体行为"""
if len(description) < 10:
raise ValueError("描述长度至少 10 个字符")
self.description = description
self.updated_at = datetime.now()
def __eq__(self, other) -> bool:
"""实体通过 ID 判断相等性"""
if not isinstance(other, Product):
return False
return self.id == other.id
def __hash__(self) -> int:
"""实体哈希基于 ID"""
return hash(self.id)
实体的关键特征：
- 唯一标识：每个实体有唯一的 ID
- 可变状态：实体状态可以随时间改变
- 业务行为：实体封装相关业务逻辑
- 生命周期：实体有创建、修改、销毁等生命周期
2.2 值对象：描述特征的不变对象
值对象没有唯一标识，通过属性值定义，通常是不可变的。
from dataclasses import dataclass
from typing import Tuple
@dataclass(frozen=True)
class Money:
"""货币值对象 - 不可变"""
amount: float
currency: str = "CNY"
def __post_init__(self):
"""验证逻辑"""
if self.amount < 0:
raise ValueError("金额不能为负数")
if len(self.currency) != 3:
raise ValueError("货币代码必须是 3 个字符")
def add(self, other: 'Money') -> 'Money':
"""加法运算"""
if self.currency != other.currency:
raise ValueError("货币类型不匹配")
return Money(self.amount + other.amount, self.currency)
def multiply(self, multiplier: float) -> 'Money':
"""乘法运算"""
return Money(self.amount * multiplier, self.currency)
@classmethod
def zero(cls, currency: str = "CNY") -> 'Money':
"""工厂方法 - 创建零值"""
return Money(0.0, currency)
@dataclass(frozen=True)
class Address:
"""地址值对象 - 不可变"""
street: str
city: str
state: str
postal_code: str
country: str
def __post_init__(self):
"""地址验证逻辑"""
if not self.postal_code.isdigit():
raise ValueError("邮政编码必须是数字")
def get_full_address(self) -> str:
"""获取完整地址"""
return f"{self.street}, {self.city}, {self.state} {self.postal_code}, {self.country}"
值对象的设计原则：
- 不可变性：创建后状态不能改变
- 无标识性：通过属性值区分
- 自包含验证：创建时进行有效性验证
- 副作用自由：操作产生新对象而不修改原有对象
2.3 聚合：一致性边界的设计
聚合是一组相关对象的集合，具有根实体和边界，保证业务规则的一致性。
from dataclasses import dataclass, field
from typing import List, Set
from uuid import UUID, uuid4
@dataclass
class OrderItem:
"""订单项实体"""
product_id: UUID
product_name: str
price: float
quantity: int
@property
def subtotal(self) -> float:
"""计算小计"""
return self.price * self.quantity
def change_quantity(self, new_quantity: int) -> None:
"""修改数量"""
if new_quantity <= 0:
raise ValueError("数量必须大于 0")
self.quantity = new_quantity
@dataclass
class Order:
"""订单聚合根"""
id: UUID = field(default_factory=uuid4)
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "created"
created_at: datetime = field(default_factory=datetime.now)
# 聚合内部业务规则
MAX_ITEMS = 50
MIN_TOTAL = 0.01
def add_item(self, product_id: UUID, product_name: str, price: float, quantity: int) -> None:
"""添加订单项"""
if self.status != "created":
raise ValueError("只能修改处于创建状态的订单")
if len(self.items) >= self.MAX_ITEMS:
raise ValueError(f"订单最多包含{self.MAX_ITEMS}个商品")
# 检查是否已存在相同商品
for item in self.items:
if item.product_id == product_id:
item.change_quantity(item.quantity + quantity)
return
# 添加新项
new_item = OrderItem(product_id, product_name, price, quantity)
self.items.append(new_item)
def remove_item(self, product_id: UUID) -> None:
"""移除订单项"""
if self.status != "created":
raise ValueError("只能修改处于创建状态的订单")
self.items = [item for item in self.items if item.product_id != product_id]
@property
def total_amount(self) -> float:
"""计算总金额"""
return sum(item.subtotal for item in self.items)
def place_order(self) -> None:
"""提交订单"""
if self.status != "created":
raise ValueError("订单状态不正确")
if not self.items:
raise ValueError("订单不能为空")
if self.total_amount < self.MIN_TOTAL:
raise ValueError("订单金额太小")
self.status = "placed"
def cancel(self) -> None:
"""取消订单"""
if self.status not in ["created", "placed"]:
raise ValueError("订单无法取消")
self.status = "cancelled"
class OrderAggregate:
"""订单聚合 - 封装聚合根和业务规则"""
def __init__(self, order: Order):
self.order = order
def apply_discount(self, discount_percentage: float) -> None:
"""应用折扣"""
if not 0 < discount_percentage <= 100:
raise ValueError("折扣比例必须在 0-100 之间")
for item in self.order.items:
original_price = item.price
discounted_price = original_price * (1 - discount_percentage / 100)
item.price = round(discounted_price, 2)
聚合设计的关键要点：
- 明确的边界：定义哪些对象属于聚合内部
- 根实体：通过根实体访问内部对象
- 不变条件：聚合始终保持业务规则一致性
- 事务边界：聚合是持久化的工作单元
2.4 仓储：领域对象的持久化抽象
仓储提供领域对象的持久化抽象，使领域层不依赖具体的数据存储技术。
from abc import ABC, abstractmethod
from typing import List, Optional
from uuid import UUID
class OrderRepository(ABC):
"""订单仓储接口 - 领域层定义"""
@abstractmethod
def save(self, order: Order) -> None:
"""保存订单"""
pass
@abstractmethod
def find_by_id(self, order_id: UUID) -> Optional[Order]:
"""根据 ID 查找订单"""
pass
@abstractmethod
def find_by_customer_id(self, customer_id: UUID) -> List[Order]:
"""根据客户 ID 查找订单"""
pass
@abstractmethod
def find_pending_orders(self) -> List[Order]:
"""查找待处理订单"""
pass
@abstractmethod
def delete(self, order_id: UUID) -> bool:
"""删除订单"""
pass
class SQLAlchemyOrderRepository(OrderRepository):
"""SQLAlchemy 实现的订单仓储"""
def __init__(self, session_factory):
self.session_factory = session_factory
def save(self, order: Order) -> None:
"""保存订单到数据库"""
with self.session_factory() as session:
# 转换为持久化对象
order_entity = self._to_entity(order)
session.merge(order_entity)
session.commit()
def find_by_id(self, order_id: UUID) -> Optional[Order]:
"""从数据库查找订单"""
with self.session_factory() as session:
from infrastructure.orm import OrderEntity
# 延迟导入避免循环依赖
entity = session.query(OrderEntity).filter_by(id=order_id).first()
return self._to_domain(entity) if entity else None
def _to_entity(self, order: Order):
"""领域对象转换为持久化对象"""
# 实现转换逻辑
pass
def _to_domain(self, entity) -> Order:
"""持久化对象转换为领域对象"""
# 实现转换逻辑
pass
class InMemoryOrderRepository(OrderRepository):
"""内存实现的订单仓储 - 用于测试"""
def __init__(self):
self._orders: dict[UUID, Order] = {}
def save(self, order: Order) -> None:
"""保存到内存"""
self._orders[order.id] = order
def find_by_id(self, order_id: UUID) -> Optional[Order]:
"""从内存查找"""
return self._orders.get(order_id)
def find_by_customer_id(self, customer_id: UUID) -> List[Order]:
"""根据客户 ID 查找"""
return [order for order in self._orders.values() if order.customer_id == customer_id]
def find_pending_orders(self) -> List[Order]:
"""查找待处理订单"""
return [order for order in self._orders.values() if order.status in ['created', 'placed']]
def delete(self, order_id: UUID) -> bool:
"""删除订单"""
if order_id in self._orders:
del self._orders[order_id]
return True
return False
仓储模式的优势：
- 持久化抽象：领域层不关心数据存储细节
- 测试友好：可以轻松替换为内存实现
- 查询封装：集中管理所有数据查询逻辑
- 工作单元模式：支持事务管理
3 项目架构与分层设计
3.1 标准的 DDD 分层架构
一个良好的 DDD 项目结构应该清晰分离关注点，下图展示了标准的 DDD 分层架构：
对应的 Python 项目结构：
3.2 应用层实现
应用层负责协调领域对象，处理用例流程。
from typing import List
from uuid import UUID
class OrderDTO:
"""订单数据传输对象"""
def __init__(self, order_id: UUID, customer_id: UUID, total: float, status: str):
self.order_id = order_id
self.customer_id = customer_id
self.total = total
self.status = status
@classmethod
def from_domain(cls, order: 'Order') -> 'OrderDTO':
"""从领域对象创建 DTO"""
return cls(
order_id=order.id,
customer_id=order.customer_id,
total=order.total_amount,
status=order.status
)
class PlaceOrderCommand:
"""下单命令"""
def __init__(self, customer_id: UUID, items: List[dict]):
self.customer_id = customer_id
self.items = items # [{product_id, quantity}]
class OrderApplicationService:
"""订单应用服务"""
def __init__(self, order_repo: OrderRepository, product_repo: 'ProductRepository'):
self.order_repo = order_repo
self.product_repo = product_repo
def place_order(self, command: PlaceOrderCommand) -> OrderDTO:
"""处理下单用例"""
# 1. 创建订单聚合
order = Order(customer_id=command.customer_id)
# 2. 添加订单项
for item in command.items:
product = self.product_repo.find_by_id(item['product_id'])
if not product:
raise ValueError(f"产品不存在：{item['product_id']}")
order.add_item(
product_id=product.id,
product_name=product.name,
price=product.price,
quantity=item['quantity']
)
# 3. 提交订单
order.place_order()
# 4. 保存订单
self.order_repo.save(order)
# 5. 返回 DTO
return OrderDTO.from_domain(order)
def get_order_details(self, order_id: UUID) -> OrderDTO:
"""获取订单详情"""
order = self.order_repo.find_by_id(order_id)
if not order:
raise ValueError("订单不存在")
return OrderDTO.from_domain(order)
def cancel_order(self, order_id: UUID) -> None:
"""取消订单"""
order = self.order_repo.find_by_id(order_id)
if not order:
raise ValueError("订单不存在")
order.cancel()
self.order_repo.save(order)
4 Pydantic 在 DDD 中的高级实践
4.1 Pydantic 值对象实现
Pydantic v2 提供了强大的数据验证功能，非常适合实现值对象。
from pydantic import BaseModel, Field, validator
from typing import Literal
class Email(BaseModel):
"""邮箱值对象 - 使用 Pydantic 验证"""
value: str
@validator('value')
def validate_email(cls, v):
if '@' not in v:
raise ValueError('无效的邮箱地址')
return v.lower()
def __str__(self):
return self.value
def __eq__(self, other):
if not isinstance(other, Email):
return False
return self.value == other.value
class PhoneNumber(BaseModel):
"""电话号码值对象"""
country_code: str = Field(default="+86", regex=r'^\+\d{1,3}$')
number: str = Field(..., regex=r'^\d{10,11}$')
@property
def full_number(self) -> str:
return f"{self.country_code}{self.number}"
@classmethod
def create_chinese_number(cls, number: str) -> 'PhoneNumber':
return cls(country_code="+86", number=number)
class AddressVO(BaseModel):
"""地址值对象 - 使用 Pydantic 高级特性"""
street: str = Field(..., min_length=1, max_length=100)
city: str = Field(..., min_length=1, max_length=50)
state: str = Field(..., min_length=1, max_length=50)
postal_code: str = Field(..., regex=r'^\d{6}$')
country: str = Field(default="中国")
class Config:
frozen = True # 不可变对象
def get_formatted_address(self) -> str:
return f"{self.country} {self.state} {self.city} {self.street} {self.postal_code}"
# 使用 Pydantic 的区分器实现多态值对象
class PaymentMethod(BaseModel):
"""支付方式基类"""
type: str
class Config:
frozen = True
class CreditCardPayment(PaymentMethod):
"""信用卡支付"""
type: Literal['credit_card'] = 'credit_card'
card_number: str = Field(..., regex=r'^\d{16}$')
expiry_date: str = Field(..., regex=r'^\d{2}/\d{2}$')
cvv: str = Field(..., regex=r'^\d{3}$')
class AlipayPayment(PaymentMethod):
"""支付宝支付"""
type: Literal['alipay'] = 'alipay'
account: str = Field(..., min_length=5)
qr_code: str = Field(None)
from typing import Annotated, Union
from pydantic import Field
# 使用区分器联合类型
Payment = Annotated[
Union[CreditCardPayment, AlipayPayment],
Field(discriminator='type')
]
4.2 领域事件与 Pydantic 集成
领域事件是 DDD 中实现模块间解耦的重要机制。
from pydantic import BaseModel
from datetime import datetime
from uuid import UUID, uuid4
from typing import Type, Dict, TypeVar
class DomainEvent(BaseModel):
"""领域事件基类"""
event_id: UUID = Field(default_factory=uuid4)
occurred_on: datetime = Field(default_factory=datetime.now)
event_type: str = Field(...)
class Config:
frozen = True # 事件不可变
def __init__(self, **kwargs):
if 'event_type' not in kwargs:
kwargs['event_type'] = self.__class__.__name__
super().__init__(**kwargs)
class OrderCreatedEvent(DomainEvent):
"""订单创建事件"""
order_id: UUID
customer_id: UUID
total_amount: float
items: list
class OrderPaidEvent(DomainEvent):
"""订单支付事件"""
order_id: UUID
payment_id: UUID
paid_amount: float
payment_method: str
class OrderCancelledEvent(DomainEvent):
"""订单取消事件"""
order_id: UUID
reason: str
cancelled_by: UUID
class EventBus:
"""简单事件总线"""
def __init__(self):
self._handlers: Dict[Type[DomainEvent], list] = {}
def subscribe(self, event_type: Type[DomainEvent], handler: callable):
"""订阅事件"""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
"""发布事件"""
event_type = type(event)
if event_type in self._handlers:
for handler in self._handlers[event_type]:
try:
handler(event)
except Exception as e:
print(f"事件处理失败：{e}")
# 事件处理示例
def send_order_confirmation_email(event: OrderCreatedEvent):
"""发送订单确认邮件"""
print(f"发送订单确认邮件：{event.order_id}")
def update_inventory(event: OrderCreatedEvent):
"""更新库存"""
print(f"更新库存 for order: {event.order_id}")
def process_payment_notification(event: OrderPaidEvent):
"""处理支付通知"""
print(f"处理支付：{event.payment_id} for order: {event.order_id}")
# 使用事件总线
event_bus = EventBus()
event_bus.subscribe(OrderCreatedEvent, send_order_confirmation_email)
event_bus.subscribe(OrderCreatedEvent, update_inventory)
event_bus.subscribe(OrderPaidEvent, process_payment_notification)
5 企业级实战案例：电商系统完整实现
5.1 领域模型设计
下面通过一个完整的电商系统案例展示 DDD 在实际项目中的应用。
from datetime import datetime
from enum import Enum
from typing import List, Optional
from uuid import UUID, uuid4
class OrderStatus(Enum):
CREATED = "created"
PLACED = "placed"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
class ProductStatus(Enum):
ACTIVE = "active"
INACTIVE = "inactive"
OUT_OF_STOCK = "out_of_stock"
@dataclass
class Customer:
"""客户实体"""
id: UUID
name: str
email: str
phone: str
created_at: datetime
status: str = "active"
def change_email(self, new_email: str) -> None:
"""修改邮箱"""
# 邮箱验证逻辑
self.email = new_email
def is_eligible_for_credit(self) -> bool:
"""判断是否有信用资格"""
# 复杂的业务规则
return True
class CatalogService:
"""目录领域服务"""
def __init__(self, product_repo: 'ProductRepository'):
self.product_repo = product_repo
def search_products(self, query: str, category: str = None) -> List['Product']:
"""搜索产品"""
products = self.product_repo.find_active()
filtered = [p for p in products if query.lower() in p.name.lower()]
if category:
filtered = [p for p in filtered if p.category == category]
return sorted(filtered, key=lambda x: x.name)
def calculate_discount(self, product: 'Product', customer: Customer) -> float:
"""计算产品折扣"""
base_discount = 0.0
# VIP 客户折扣
if customer.is_eligible_for_credit():
base_discount += 0.1
# 促销折扣
if product.is_on_promotion():
base_discount += 0.05
return min(base_discount, 0.3) # 最大折扣 30%
class PricingService:
"""定价领域服务"""
def calculate_total(self, order: Order) -> float:
"""计算订单总价"""
base_total = order.total_amount
# 应用折扣规则
discounts = self._calculate_discounts(order)
# 应用税费
tax = self._calculate_tax(order)
return base_total - discounts + tax
def _calculate_discounts(self, order: Order) -> float:
"""计算折扣"""
# 复杂的折扣逻辑
return 0.0
def _calculate_tax(self, order: Order) -> float:
"""计算税费"""
# 复杂的税费逻辑
return order.total_amount * 0.1
5.2 完整的应用服务实现
class ECommerceApplicationService:
"""电商应用服务"""
def __init__(self, order_repo: OrderRepository, product_repo: 'ProductRepository', customer_repo: 'CustomerRepository', event_bus: EventBus):
self.order_repo = order_repo
self.product_repo = product_repo
self.customer_repo = customer_repo
self.event_bus = event_bus
self.catalog_service = CatalogService(product_repo)
self.pricing_service = PricingService()
def create_order(self, command: 'CreateOrderCommand') -> OrderDTO:
"""创建订单用例"""
# 1. 验证客户
customer = self.customer_repo.find_by_id(command.customer_id)
if not customer:
raise ValueError("客户不存在")
# 2. 创建订单
order = Order(customer_id=customer.id)
# 3. 添加商品
for item in command.items:
product = self.product_repo.find_by_id(item.product_id)
if not product or product.status != ProductStatus.ACTIVE:
raise ValueError(f"商品不可用：{item.product_id}")
if product.stock_quantity < item.quantity:
raise ValueError(f"商品库存不足：{product.name}")
order.add_item(
product_id=product.id,
product_name=product.name,
price=product.price,
quantity=item.quantity
)
# 4. 计算价格
final_total = self.pricing_service.calculate_total(order)
# 这里可以设置最终价格
# 5. 提交订单
order.place_order()
# 6. 保存订单
self.order_repo.save(order)
# 7. 发布领域事件
event = OrderCreatedEvent(
order_id=order.id,
customer_id=order.customer_id,
total_amount=order.total_amount,
items=[{'product_id': item.product_id, 'quantity': item.quantity} for item in order.items]
)
self.event_bus.publish(event)
# 8. 返回结果
return OrderDTO.from_domain(order)
def process_payment(self, command: 'ProcessPaymentCommand') -> None:
"""处理支付用例"""
order = self.order_repo.find_by_id(command.order_id)
if not order:
raise ValueError("订单不存在")
# 处理支付逻辑
payment_success = True # 模拟支付成功
if payment_success:
order.mark_as_paid()
self.order_repo.save(order)
# 发布支付事件
event = OrderPaidEvent(
order_id=order.id,
payment_id=command.payment_id,
paid_amount=command.amount,
payment_method=command.method
)
self.event_bus.publish(event)
class CreateOrderCommand:
"""创建订单命令"""
def __init__(self, customer_id: UUID, items: List['OrderItemCommand']):
self.customer_id = customer_id
self.items = items
class OrderItemCommand:
"""订单项命令"""
def __init__(self, product_id: UUID, quantity: int):
self.product_id = product_id
self.quantity = int(quantity)
下面的序列图展示了订单创建过程的完整流程：
6 性能优化与最佳实践
6.1 DDD 性能优化策略
虽然 DDD 提供了良好的架构，但在性能敏感的场景下需要特别考虑。
from functools import lru_cache
from typing import Generic, TypeVar
import time
T = TypeVar('T')
class CachedRepository(Generic[T]):
"""带缓存的仓储实现"""
def __init__(self, underlying_repo, ttl_seconds: int = 300):
self.underlying_repo = underlying_repo
self.ttl = ttl_seconds
self._cache = {}
self._cache_timestamps = {}
def find_by_id(self, entity_id: UUID) -> Optional[T]:
"""带缓存的查询"""
cache_key = str(entity_id)
# 检查缓存是否存在且未过期
if cache_key in self._cache:
if time.time() - self._cache_timestamps[cache_key] < self.ttl:
return self._cache[cache_key]
else:
# 缓存过期，清除
del self._cache[cache_key]
del self._cache_timestamps[cache_key]
# 从底层仓储加载
entity = self.underlying_repo.find_by_id(entity_id)
if entity:
self._cache[cache_key] = entity
self._cache_timestamps[cache_key] = time.time()
return entity
def save(self, entity: T) -> None:
"""保存并更新缓存"""
self.underlying_repo.save(entity)
# 更新缓存
cache_key = str(entity.id)
self._cache[cache_key] = entity
self._cache_timestamps[cache_key] = time.time()
def invalidate_cache(self, entity_id: UUID) -> None:
"""使缓存失效"""
cache_key = str(entity_id)
if cache_key in self._cache:
del self._cache[cache_key]
del self._cache_timestamps[cache_key]
class ReadModelService:
"""查询模型服务 - CQRS 模式"""
def __init__(self, database_session):
self.session = database_session
@lru_cache(maxsize=1000)
def get_order_summary(self, order_id: UUID) -> dict:
"""获取订单摘要 - 使用缓存"""
# 直接查询读取模型，避免聚合加载
query = """
SELECT o.id, o.status, o.total_amount, c.name as customer_name
FROM order_read_model o
JOIN customers c ON o.customer_id = c.id
WHERE o.id = :order_id
"""
result = self.session.execute(query, {'order_id': str(order_id)}).fetchone()
return dict(result) if result else None
def get_customer_orders(self, customer_id: UUID, page: int = 1, size: int = 20) -> dict:
"""分页查询客户订单"""
offset = (page - 1) * size
query = """
SELECT id, status, total_amount, created_at
FROM order_read_model
WHERE customer_id = :customer_id
ORDER BY created_at DESC
LIMIT :limit OFFSET :offset
"""
results = self.session.execute(query, {
'customer_id': str(customer_id),
'limit': size,
'offset': offset
}).fetchall()
total_query = "SELECT COUNT(*) FROM order_read_model WHERE customer_id = :customer_id"
total = self.session.execute(total_query, {
'customer_id': str(customer_id)
}).scalar()
return {
'items': [dict(row) for row in results],
'total': total,
'page': page,
'size': size
}
6.2 测试策略
DDD 架构的测试应该分层进行，确保每层的职责清晰。
import pytest
from unittest.mock import Mock, create_autospec
class TestOrderAggregate:
"""订单聚合测试"""
def test_create_order(self):
"""测试订单创建"""
order = Order(customer_id=uuid4())
assert order.status == OrderStatus.CREATED
assert len(order.items) == 0
def test_add_item(self):
"""测试添加订单项"""
order = Order(customer_id=uuid4())
product_id = uuid4()
order.add_item(product_id, "测试产品", 100.0, 2)
assert len(order.items) == 1
assert order.items[0].product_id == product_id
assert order.items[0].quantity == 2
def test_place_order_valid(self):
"""测试有效的订单提交"""
order = Order(customer_id=uuid4())
order.add_item(uuid4(), "测试产品", 100.0, 1)
order.place_order()
assert order.status == OrderStatus.PLACED
def test_place_order_empty(self):
"""测试空订单提交"""
order = Order(customer_id=uuid4())
with pytest.raises(ValueError, match="订单不能为空"):
order.place_order()
class TestOrderApplicationService:
"""订单应用服务测试"""
@pytest.fixture
def service(self):
"""创建测试服务"""
order_repo = create_autospec(OrderRepository)
product_repo = create_autospec(ProductRepository)
customer_repo = create_autospec(CustomerRepository)
event_bus = create_autospec(EventBus)
return ECommerceApplicationService(
order_repo, product_repo, customer_repo, event_bus
)
def test_create_order_success(self, service):
"""测试成功创建订单"""
# 设置 mock 行为
customer_id = uuid4()
product_id = uuid4()
service.customer_repo.find_by_id.return_value = Customer(
id=customer_id,
name="测试客户",
email="[email protected]",
phone="13800138000",
created_at=datetime.now()
)
service.product_repo.find_by_id.return_value = Product(
id=product_id,
name="测试产品",
price=100.0,
status=ProductStatus.ACTIVE
)
# 执行测试
command = CreateOrderCommand(
customer_id=customer_id,
items=[OrderItemCommand(product_id=product_id, quantity=1)]
)
result = service.create_order(command)
# 验证结果
assert result is not None
service.order_repo.save.assert_called_once()
service.event_bus.publish.assert_called_once()
# 集成测试
@pytest.mark.integration
class TestOrderIntegration:
"""订单集成测试"""
def test_complete_order_flow(self, database_session):
"""完整订单流程测试"""
# 使用测试数据库
repo = SQLAlchemyOrderRepository(database_session)
# 创建测试数据
customer_id = uuid4()
product_id = uuid4()
# 执行订单流程
order = Order(customer_id=customer_id)
order.add_item(product_id, "测试产品", 100.0, 1)
order.place_order()
repo.save(order)
# 验证持久化
saved_order = repo.find_by_id(order.id)
assert saved_order is not None
assert saved_order.status == OrderStatus.PLACED
7 常见问题与解决方案
7.1 DDD 实施中的典型挑战
问题 1：领域模型与持久化模型的冲突
解决方案：使用抗腐蚀层和映射层
class OrderMapper:
"""订单映射器 - 解决领域模型与持久化模型的差异"""
@staticmethod
def to_entity(domain_order: Order) -> 'OrderEntity':
"""领域对象转换为持久化对象"""
entity = OrderEntity()
entity.id = domain_order.id
entity.customer_id = domain_order.customer_id
entity.status = domain_order.status.value # 枚举转字符串
entity.total_amount = domain_order.total_amount
entity.created_at = domain_order.created_at
# 转换订单项
entity.items = [
OrderItemEntity(
product_id=item.product_id,
product_name=item.product_name,
price=item.price,
quantity=item.quantity
)
for item in domain_order.items
]
return entity
@staticmethod
def to_domain(entity: 'OrderEntity') -> Order:
"""持久化对象转换为领域对象"""
order = Order(
id=entity.id,
customer_id=entity.customer_id,
status=OrderStatus(entity.status), # 字符串转枚举
created_at=entity.created_at
)
# 重建订单项
for item_entity in entity.items:
order_item = OrderItem(
product_id=item_entity.product_id,
product_name=item_entity.product_name,
price=item_entity.price,
quantity=item_entity.quantity
)
order.items.append(order_item)
return order
问题 2：复杂查询的性能问题
解决方案：CQRS 模式
class OrderQueryService:
"""订单查询服务 - CQRS 查询端"""
def __init__(self, read_db_session):
self.session = read_db_session
def search_orders(self, filters: dict) -> dict:
"""复杂订单查询"""
query = self._build_query(filters)
results = self.session.execute(query).fetchall()
return {
'orders': [dict(row) for row in results],
'total': len(results)
}
def _build_query(self, filters: dict) -> tuple:
"""构建查询语句"""
base_query = """
SELECT o.*, c.name as customer_name
FROM order_read_model o
JOIN customers c ON o.customer_id = c.id
WHERE 1=1
"""
conditions = []
params = {}
if 'status' in filters:
conditions.append("o.status = :status")
params['status'] = filters['status']
if 'customer_id' in filters:
conditions.append("o.customer_id = :customer_id")
params['customer_id'] = filters['customer_id']
if conditions:
base_query += " AND " + " AND ".join(conditions)
base_query += " ORDER BY o.created_at DESC"
return base_query, params
8 总结与展望
领域驱动设计在 Python 中的实施已经相当成熟，通过本文的完整实践指南，可以看到 Python 生态为 DDD 提供了优秀的支持。随着 Python 在 AI、数据科学等领域的持续发展，DDD 将在复杂系统架构中发挥更加重要的作用。
8.1 关键收获
- Python 与 DDD 高度契合：动态特性让领域模型表达更自然
- 分层架构至关重要：清晰的职责分离是成功的关键
- Pydantic 是强大工具：在值对象和验证场景中表现卓越
- 测试驱动开发：DDD 架构天然支持高质量的测试策略
8.2 未来趋势
- AI 增强的领域建模：LLM 可以帮助识别领域概念和关系
- 微服务与 DDD 深度结合：界限上下文自然对应微服务边界
- 实时系统设计：事件驱动架构与 DDD 完美配合
- 开发工具完善：更多专为 Python DDD 设计的工具和框架
官方文档与权威参考
DDD 不是银弹，但在复杂业务系统中，它提供了应对复杂性的有效方法。希望本文能帮助你在 Python 项目中成功实施领域驱动设计，构建出可维护、可扩展的高质量软件系统。
