本指南通过构建“电商订单系统”案例展示如何在 Python 中实现业务逻辑与外部依赖Web 框架、数据库 ORM、消息队列等的完全解耦提升可测试性、可维护性与扩展性。一、架构核心原则依赖规则外层依赖内层内层Domain绝不依赖外层Adapters/Infrastructure。领域纯净核心业务逻辑不依赖任何外部技术实现Flask、SQLAlchemy、Redis 等。端口与适配器通过抽象接口Port隔离外部依赖由适配器Adapter实现具体逻辑。领域模型充血业务规则和行为封装在领域模型内部避免贫血模型。二、项目结构project/ ├── domain/ # 核心领域层无外部依赖 │ ├── models.py # 领域模型包含业务行为 │ └── ports.py # 端口接口仓储、服务接口 ├── application/ # 应用层编排用例仅依赖 domain │ └── order_service.py # 用例实现 ├── adapters/ # 适配器层实现端口依赖外部技术 │ ├── web/ │ │ └── flask_routes.py │ └── persistence/ │ ├── orm_models.py # ORM 模型独立于领域模型 │ └── sqlalchemy_repo.py ├── infrastructure/ # 基础设施层配置、组装 │ └── config.py └── tests/ ├── unit/ # 纯内存/Mock 测试 └── integration/ # 真实组件集成测试三、核心领域模型充血模型将业务规则、校验和 ID 生成策略封装在领域对象内部而非散落在 Service 中。# domain/models.py import uuid from datetime import datetime, timezone from dataclasses import dataclass, field from decimal import Decimal dataclass(frozenTrue) class OrderItem: sku: str quantity: int 1 dataclass class Order: user_id: str total_amount: Decimal items: list[OrderItem] order_id: str field(default_factorylambda: str(uuid.uuid4())) status: str pending created_at: datetime field(default_factorylambda: datetime.now(timezone.utc)) classmethod def create(cls, user_id: str, total_amount: Decimal, items: list[dict]) - Order: 工厂方法封装创建逻辑与业务校验 if total_amount 0: raise ValueError(订单金额必须大于0) if not items: raise ValueError(订单至少包含一个商品) order_items [OrderItem(skui[sku], quantityi.get(quantity, 1)) for i in items] return cls(user_iduser_id, total_amounttotal_amount, itemsorder_items) def confirm(self): 状态流转业务规则 if self.status ! pending: raise ValueError(f只有待处理订单可确认当前状态: {self.status}) self.status confirmed四、端口定义接口契约应用层和领域层只依赖这些抽象接口不关心具体实现。# domain/ports.py from abc import ABC, abstractmethod from typing import Optional from domain.models import Order class OrderRepository(ABC): 仓储端口定义数据访问契约 abstractmethod def save(self, order: Order) - Order: ... abstractmethod def find_by_id(self, order_id: str) - Optional[Order]: ... class EventBus(ABC): 事件总线端口 abstractmethod def publish(self, event_name: str, payload: dict) - None: ...五、应用层用例编排⚠️关键原则应用层只导入 domain 层绝不导入 adapters 层。# application/order_service.py from decimal import Decimal from domain.models import Order from domain.ports import OrderRepository, EventBus class CreateOrderUseCase: def __init__(self, repo: OrderRepository, event_bus: EventBus | None None): self.repo repo self.event_bus event_bus def execute(self, user_id: str, total_amount: Decimal, items: list[dict]) - Order: # 1. 调用领域模型工厂方法业务校验在此完成 order Order.create(user_iduser_id, total_amounttotal_amount, itemsitems) # 2. 持久化 saved_order self.repo.save(order) # 3. 发布领域事件可选 if self.event_bus: self.event_bus.publish(order.created, {order_id: saved_order.order_id}) return saved_order六、适配器实现6.1 ORM 模型独立于领域模型# adapters/persistence/orm_models.py from datetime import datetime from sqlalchemy import String, Float, DateTime from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class OrderModel(Base): ORM 模型仅负责数据库映射不含业务逻辑 __tablename__ orders id: Mapped[str] mapped_column(String(36), primary_keyTrue) user_id: Mapped[str] total_amount: Mapped[float] mapped_column(Float) status: Mapped[str] created_at: Mapped[datetime] mapped_column(DateTime)6.2 仓储适配器含双向转换# adapters/persistence/sqlalchemy_repo.py from sqlalchemy.orm import Session from domain.models import Order, OrderItem from domain.ports import OrderRepository from .orm_models import OrderModel class SqlAlchemyOrderRepository(OrderRepository): def __init__(self, session: Session): self.session session def _to_domain(self, model: OrderModel) - Order: ORM → Domain 转换 from decimal import Decimal return Order( order_idmodel.id, user_idmodel.user_id, total_amountDecimal(str(model.total_amount)), statusmodel.status, created_atmodel.created_at, items[] # 简化示例实际应关联查询 OrderItem ) def _to_model(self, domain: Order) - OrderModel: Domain → ORM 转换 return OrderModel( iddomain.order_id, user_iddomain.user_id, total_amountfloat(domain.total_amount), statusdomain.status, created_atdomain.created_at ) def save(self, order: Order) - Order: model self._to_model(order) self.session.merge(model) self.session.commit() return order def find_by_id(self, order_id: str): model self.session.query(OrderModel).filter(OrderModel.id order_id).first() return self._to_domain(model) if model else None6.3 Web 适配器# adapters/web/flask_routes.py from decimal import Decimal from flask import Flask, request, jsonify from application.order_service import CreateOrderUseCase def register_routes(app: Flask, use_case: CreateOrderUseCase): app.route(/api/orders, methods[POST]) def create_order(): data request.get_json() try: order use_case.execute( user_iddata[user_id], total_amountDecimal(str(data[total_amount])), itemsdata.get(items, []) ) return jsonify({ order_id: order.order_id, status: order.status }), 201 except ValueError as e: return jsonify({error: str(e)}), 400七、依赖注入与组装# infrastructure/config.py from flask import Flask from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from adapters.persistence.sqlalchemy_repo import SqlAlchemyOrderRepository from adapters.persistence.orm_models import Base from application.order_service import CreateOrderUseCase from adapters.web.flask_routes import register_routes def create_app(database_url: str sqlite:///orders.db) - Flask: # 1. 初始化基础设施 engine create_engine(database_url) Base.metadata.create_all(bindengine) Session sessionmaker(bindengine) # 2. 组装依赖手动 DI也可用 dependency-injector 等库 repo SqlAlchemyOrderRepository(Session()) use_case CreateOrderUseCase(reporepo) # 3. 注册适配器 app Flask(__name__) register_routes(app, use_case) return app八、测试示例8.1 单元测试纯领域逻辑零外部依赖# tests/unit/test_domain.py import pytest from decimal import Decimal from domain.models import Order def test_create_order_success(): order Order.create(user_idu1, total_amountDecimal(99.9), items[{sku: SKU-001}]) assert order.status pending assert order.user_id u1 assert len(order.items) 1 def test_create_order_invalid_amount(): with pytest.raises(ValueError, match订单金额必须大于0): Order.create(user_idu1, total_amountDecimal(-1), items[{sku: SKU-001}]) def test_confirm_order(): order Order.create(user_idu1, total_amountDecimal(50), items[{sku: A}]) order.confirm() assert order.status confirmed8.2 集成测试使用内存数据库# tests/integration/test_order_service.py import pytest from decimal import Decimal from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from adapters.persistence.orm_models import Base from adapters.persistence.sqlalchemy_repo import SqlAlchemyOrderRepository from application.order_service import CreateOrderUseCase pytest.fixture def use_case(): engine create_engine(sqlite:///:memory:) Base.metadata.create_all(bindengine) session sessionmaker(bindengine)() repo SqlAlchemyOrderRepository(session) return CreateOrderUseCase(reporepo) def test_create_and_persist_order(use_case): order use_case.execute( user_idu1, total_amountDecimal(100.0), items[{sku: SKU-001, quantity: 2}] ) assert order.order_id is not None assert order.status pending # 验证持久化 retrieved use_case.repo.find_by_id(order.order_id) assert retrieved is not None assert retrieved.total_amount Decimal(100.0)九、收益总结维度效果可测试性领域逻辑单元测试毫秒级执行无需启动 DB/Web 服务技术无关性可随时将 Flask 替换为 FastAPISQLAlchemy 替换为 Tortoise-ORM核心代码零修改业务安全所有校验和状态流转封装在领域模型中无法被绕过团队协作领域专家与开发人员可围绕domain/models.py进行 Ubiquitous Language 对齐可维护性变更影响范围可控修改数据库表结构不影响业务逻辑十、扩展方向CQRS读写分离时为 Query 定义独立的 ReadModel 和 QueryPort。事件溯源将save()替换为append_event()配合 EventStore 适配器。异步支持端口接口使用async def适配器对应使用asyncpg/httpx。多租户/插件化通过配置文件动态加载不同的适配器实现类。注意本指南适用于中大型项目或需要长期演进的系统。对于简单的 CRUD 脚本或原型验证请权衡架构复杂度避免过度设计。