Python领域驱动设计DDD

DDD是一种以领域为核心的软件设计方法，适合复杂业务系统。

DDD核心概念

战略设计层次

领域 → 子域 → 限界上下文

领域：整个业务领域
子域：核心域、支撑域、通用域
限界上下文：明确的边界，独立模型

战术设计要素

# DDD战术元素
ELEMENTS = {
    'Entity': '有唯一标识的领域对象',
    'Value Object': '无标识，由属性定义的对象',
    'Aggregate': '实体和值对象的集合，事务边界',
    'Domain Service': '不属于实体的领域行为',
    'Repository': '持久化抽象',
    'Application Service': '应用层协调',
    'Factory': '复杂对象创建',
}

实体（Entity）

实体定义

from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class UserId:
    "值对象：用户ID"
    value: UUID

class User:
    "实体：有唯一标识"
    def __init__(self, id: UserId, name: str, email: str):
        self._id = id          # 唯一标识
        self._name = name
        self._email = email

    @property
    def id(self) -> UserId:
        return self._id

    @property
    def name(self) -> str:
        return self._name

    def change_name(self, new_name: str):
        "行为方法：改变名字"
        if not new_name:
            raise ValueError("Name cannot be empty")
        self._name = new_name

    def change_email(self, new_email: str):
        "行为方法：改变邮箱"
        if not self._validate_email(new_email):
            raise ValueError("Invalid email")
        self._email = new_email

    def __eq__(self, other):
        "相等性比较：基于ID"
        if not isinstance(other, User):
            return False
        return self._id == other._id

    def _validate_email(self, email):
        return '@' in email

值对象（Value Object）

值对象定义

from dataclasses import dataclass
from typing import NewType

@dataclass(frozen=True)  # 不可变
class Money:
    "值对象：金额"
    amount: int
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")

    def add(self, other: Money) -> Money:
        "值对象运算返回新对象"
        if other.currency != self.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)

@dataclass(frozen=True)
class Email:
    "值对象：邮箱"
    address: str

    def __post_init__(self):
        if '@' not in self.address:
            raise ValueError("Invalid email format")

@dataclass(frozen=True)
class Address:
    "值对象：地址"
    street: str
    city: str
    country: str
    postal_code: str

    def __eq__(self, other):
        "相等性比较：基于所有属性"
        if not isinstance(other, Address):
            return False
        return (self.street == other.street and
                self.city == other.city and
                self.country == other.country)

聚合（Aggregate）

聚合定义

from typing import List
from uuid import UUID, uuid4

class OrderId:
    "值对象：订单ID"
    value: UUID

class OrderItem:
    "实体：订单项"
    def __init__(self, product_id: UUID, quantity: int, price: Money):
        self.product_id = product_id
        self._quantity = quantity
        self._price = price

    @property
    def quantity(self):
        return self._quantity

    @property
    def price(self):
        return self._price

    def increase_quantity(self, amount: int):
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self._quantity += amount

class Order:
    "聚合根：订单"
    def __init__(self, id: OrderId, customer_id: UUID):
        self._id = id              # 聚合根ID
        self._customer_id = customer_id
        self._items: List[OrderItem] = []
        self._status = 'draft'
        self._total: Money = Money(0, 'USD')

    @property
    def id(self):
        return self._id

    @property
    def items(self):
        return self._items.copy()  # 返回副本，防止外部修改

    def add_item(self, product_id: UUID, quantity: int, price: Money):
        "聚合行为：添加订单项"
        if self._status != 'draft':
            raise ValueError("Cannot modify confirmed order")

        item = OrderItem(product_id, quantity, price)
        self._items.append(item)
        self._recalculate_total()

    def remove_item(self, product_id: UUID):
        "聚合行为：移除订单项"
        if self._status != 'draft':
            raise ValueError("Cannot modify confirmed order")

        self._items = [i for i in self._items if i.product_id != product_id]
        self._recalculate_total()

    def confirm(self):
        "聚合行为：确认订单"
        if not self._items:
            raise ValueError("Order has no items")
        if self._status != 'draft':
            raise ValueError("Order already confirmed")
        self._status = 'confirmed'

    def _recalculate_total(self):
        "内部计算"
        total = Money(0, 'USD')
        for item in self._items:
            total = total.add(Money(item.quantity * item.price.amount, item.price.currency))
        self._total = total

领域服务（Domain Service）

领域服务定义

class PricingService:
    "领域服务：定价计算"

    def calculate_discount(self, customer: Customer, order: Order) -> Money:
        "不属于实体或值对象的领域行为"
        discount_rate = self._get_discount_rate(customer)

        if customer.is_premium:
            discount_rate += 0.05

        if order.total.amount > 1000:
            discount_rate += 0.02

        discount_amount = int(order.total.amount * discount_rate)
        return Money(discount_amount, order.total.currency)

    def _get_discount_rate(self, customer):
        return 0.1 if customer.order_count > 10 else 0

class TransferService:
    "领域服务：账户转账"

    def transfer(self, from_account: Account, to_account: Account, amount: Money):
        "涉及多个聚合的操作"
        if from_account.balance.currency != amount.currency:
            raise ValueError("Currency mismatch")

        if from_account.balance.amount < amount.amount:
            raise ValueError("Insufficient balance")

        from_account.debit(amount)
        to_account.credit(amount)

仓库（Repository）

仓库接口定义

from abc import ABC, abstractmethod
from typing import Optional, List

class OrderRepository(ABC):
    "仓库抽象接口"

    @abstractmethod
    def find_by_id(self, id: OrderId) -> Optional[Order]:
        "根据ID查找"
        pass

    @abstractmethod
    def find_by_customer(self, customer_id: UUID) -> List[Order]:
        "根据客户查找"
        pass

    @abstractmethod
    def save(self, order: Order):
        "保存聚合"
        pass

    @abstractmethod
    def delete(self, order: Order):
        "删除聚合"
        pass

# 具体实现（基础设施层）
class SQLOrderRepository(OrderRepository):
    def __init__(self, db_session):
        self.session = db_session

    def find_by_id(self, id: OrderId) -> Optional[Order]:
        record = self.session.query(OrderRecord).filter_by(id=id.value).first()
        if record:
            return self._to_domain(record)
        return None

    def save(self, order: Order):
        record = self._to_record(order)
        self.session.add(record)
        self.session.commit()

    def _to_domain(self, record) -> Order:
        "ORM到领域对象映射"
        order = Order(OrderId(record.id), record.customer_id)
        for item in record.items:
            order.add_item(item.product_id, item.quantity, Money(item.price, 'USD'))
        return order

    def _to_record(self, order: Order) -> OrderRecord:
        "领域对象到ORM映射"
        record = OrderRecord(
            id=order.id.value,
            customer_id=order._customer_id,
            status=order._status
        )
        for item in order._items:
            record.items.append(OrderItemRecord(
                product_id=item.product_id,
                quantity=item.quantity,
                price=item.price.amount
            ))
        return record

应用服务（Application Service）

应用服务定义

from typing import Dict

class OrderApplicationService:
    "应用服务：协调领域对象"

    def __init__(self, order_repo: OrderRepository,
                 product_repo: ProductRepository,
                 pricing_service: PricingService):
        self.order_repo = order_repo
        self.product_repo = product_repo
        self.pricing_service = pricing_service

    def create_order(self, customer_id: UUID, items: List[Dict]) -> OrderId:
        "创建订单用例"
        order = Order(OrderId(uuid4()), customer_id)

        for item_data in items:
            product = self.product_repo.find_by_id(item_data['product_id'])
            if not product:
                raise ValueError(f"Product {item_data['product_id']} not found")

            order.add_item(
                product.id,
                item_data['quantity'],
                product.price
            )

        self.order_repo.save(order)
        return order.id

    def confirm_order(self, order_id: OrderId, customer: Customer):
        "确认订单用例"
        order = self.order_repo.find_by_id(order_id)
        if not order:
            raise ValueError("Order not found")

        # 使用领域服务计算折扣
        discount = self.pricing_service.calculate_discount(customer, order)

        order.confirm()
        self.order_repo.save(order)

        return {'order_id': order.id, 'discount': discount}

DDD分层架构

# 四层架构

# 1. 用户接口层（Interface Layer）
# - 控制器/API
# - 视图/DTO
# - 请求响应处理

# api/orders.py
class OrderAPI:
    def create_order(self, request):
        dto = CreateOrderDTO.from_request(request)
        order_id = self.app_service.create_order(dto.customer_id, dto.items)
        return OrderResponseDTO(order_id)

# 2. 应用层（Application Layer）
# - 应用服务
# - 用例协调
# - 事务管理

# application/order_service.py
class OrderApplicationService:
    def create_order(self, ...):
        pass

# 3. 领域层（Domain Layer）
# - 实体
# - 值对象
# - 聚合
# - 领域服务
# - 仓库接口

# domain/order.py
class Order:
    pass

# domain/repository.py
class OrderRepository(ABC):
    pass

# 4. 基础设施层（Infrastructure Layer）
# - 仓库实现
# - ORM映射
# - 外部服务

# infrastructure/order_repository.py
class SQLOrderRepository(OrderRepository):
    pass

项目结构示例

project/
├── domain/                     # 领域层
│   ├── order/
│   │   ├── order.py           # 聚合根
│   │   ├── order_item.py      # 实体
│   │   ├── money.py           # 值对象
│   │   ├── repository.py      # 仓库接口
│   │   └── service.py         # 领域服务
│   ├── user/
│   │   ├── user.py
│   │   ├── ...
│   └── shared/                 # 共享值对象
│       ├── id.py
│       └── address.py
│
├── application/                # 应用层
│   ├── order/
│   │   ├── service.py         # 应用服务
│   │   └── dto.py             # DTO
│   └── user/
│       └── ...
│
├── infrastructure/             # 基础设施层
│   ├── persistence/
│   │   ├── order_repo.py      # 仓库实现
│   │   ├── models.py          # ORM模型
│   │   └── mapper.py          # 映射器
│   └── external/
│       ├── email_client.py
│       └── payment_gateway.py
│
└── interface/                  # 用户接口层
    ├── api/
    │   ├── order_api.py
    │   └── user_api.py
    └── dto/
        ├── order_dto.py
        └── user_dto.py

DDD实践要点

注意：DDD适合复杂业务系统，简单系统可能过度设计，根据实际情况取舍。

要点总结

• 实体：唯一标识、相等性基于ID、封装行为方法
• 值对象：不可变、属性定义相等性、运算返回新对象
• 聚合：聚合根作为入口、保证内部一致性、事务边界
• 领域服务：跨实体行为、不属于单个对象的领域逻辑
• 仓库：接口在领域层、实现在基础设施层、只操作聚合根
• 四层架构：用户接口→应用→领域→基础设施，依赖由外向内

存放路径：articles/PYTHON/专家/架构与设计/领域驱动设计DDD.md