企业官网建设流程全解析

本指南通过构建“电商订单系统”案例，展示如何在 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(frozen=True) class OrderItem: sku: str quantity: int = 1 @dataclass class Order: user_id: str total_amount: Decimal items: list[OrderItem] order_id: str = field(default_factory=lambda: str(uuid.uuid4())) status: str = "pending" created_at: datetime = field(default_factory=lambda: 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(sku=i['sku'], quantity=i.get('quantity', 1)) for i in items] return cls(user_id=user_id, total_amount=total_amount, items=order_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_id=user_id, total_amount=total_amount, items=items) # 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_key=True) 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_id=model.id, user_id=model.user_id, total_amount=Decimal(str(model.total_amount)), status=model.status, created_at=model.created_at, items=[] # 简化示例；实际应关联查询 OrderItem ) def _to_model(self, domain: Order) -> OrderModel: """Domain → ORM 转换""" return OrderModel( id=domain.order_id, user_id=domain.user_id, total_amount=float(domain.total_amount), status=domain.status, created_at=domain.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 None

6.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_id=data['user_id'], total_amount=Decimal(str(data['total_amount'])), items=data.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(bind=engine) Session = sessionmaker(bind=engine) # 2. 组装依赖（手动 DI，也可用 dependency-injector 等库） repo = SqlAlchemyOrderRepository(Session()) use_case = CreateOrderUseCase(repo=repo) # 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_id="u1", total_amount=Decimal("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_id="u1", total_amount=Decimal("-1"), items=[{"sku": "SKU-001"}]) def test_confirm_order(): order = Order.create(user_id="u1", total_amount=Decimal("50"), items=[{"sku": "A"}]) order.confirm() assert order.status == "confirmed"

8.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(bind=engine) session = sessionmaker(bind=engine)() repo = SqlAlchemyOrderRepository(session) return CreateOrderUseCase(repo=repo) def test_create_and_persist_order(use_case): order = use_case.execute( user_id="u1", total_amount=Decimal("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")

九、收益总结

十、扩展方向

• CQRS：读写分离时，为 Query 定义独立的 ReadModel 和 QueryPort。
• 事件溯源：将save()替换为append_event()，配合 EventStore 适配器。
• 异步支持：端口接口使用async def，适配器对应使用asyncpg/httpx。
• 多租户/插件化：通过配置文件动态加载不同的适配器实现类。

注意：本指南适用于中大型项目或需要长期演进的系统。对于简单的 CRUD 脚本或原型验证，请权衡架构复杂度，避免过度设计。