python-clean-architecture

Python 整洁架构

概述

用于构建 Python 服务的整洁架构参考指南。应用这些模式，将业务逻辑与框架相关内容分离，使代码易于测试、维护，并且不依赖特定框架。

项目结构

src/
├── domain/              # Core business logic — NO framework imports
│   ├── models/          # Domain entities and value objects
│   │   ├── user.py
│   │   └── order.py
│   ├── errors.py        # Domain-specific exceptions
│   └── services/        # Business logic / use cases
│       ├── user_service.py
│       └── order_service.py
├── infrastructure/      # External concerns
│   ├── repositories/    # Data access implementations
│   │   ├── user_repo.py
│   │   └── order_repo.py
│   ├── external/        # Third-party API clients
│   │   └── payment_client.py
│   └── db.py            # Database connection setup
├── api/                 # Framework layer (FastAPI, Flask, etc.)
│   ├── routers/
│   │   ├── users.py
│   │   └── orders.py
│   ├── dependencies.py  # DI wiring
│   └── error_handlers.py
└── main.py

关键规则：依赖指向内部。domain/ 不导入 infrastructure/ 或 api/ 中的任何内容。infrastructure/ 从 domain/ 导入。api/ 从两者导入。

领域模型（与 ORM 模型对比）

领域模型表示带有行为的业务概念。ORM 模型表示数据库表。应将二者分开。

# domain/models/user.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum

class UserRole(StrEnum):
    MEMBER = "member"
    ADMIN = "admin"

@dataclass
class User:
    id: str
    email: str
    name: str
    role: UserRole
    created_at: datetime
    is_active: bool = True

    def promote_to_admin(self) -> None:
        if not self.is_active:
            raise InactiveUserError(self.id)
        self.role = UserRole.ADMIN

    def deactivate(self) -> None:
        self.is_active = False

    @property
    def is_admin(self) -> bool:
        return self.role == UserRole.ADMIN


# domain/models/order.py
@dataclass
class OrderItem:
    product_id: str
    quantity: int
    unit_price: float

    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

@dataclass
class Order:
    id: str
    user_id: str
    items: list[OrderItem] = field(default_factory=list)
    status: str = "draft"

    @property
    def total(self) -> float:
        return sum(item.total for item in self.items)

    def add_item(self, product_id: str, quantity: int, unit_price: float) -> None:
        if self.status != "draft":
            raise OrderNotEditableError(self.id, self.status)
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        self.items.append(OrderItem(product_id, quantity, unit_price))

    def submit(self) -> None:
        if not self.items:
            raise EmptyOrderError(self.id)
        self.status = "submitted"

ORM 模型 — 独立关注点

ORM 模型映射数据库架构，可能包含领域模型中没有的字段（例如，password_hash）。在存储库层中映射 ORM 模型与领域模型。

# infrastructure/db_models/user_model.py
class UserModel(Base):
    __tablename__ = "users"
    id = Column(String, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    role = Column(String, nullable=False, default="member")
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, nullable=False)
    password_hash = Column(String, nullable=False)  # Not in domain model

存储库模式

在领域层中定义抽象接口。在基础设施层中实现。

# domain/repositories.py
from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    async def get_by_id(self, user_id: str) -> User | None: ...

    @abstractmethod
    async def get_by_email(self, email: str) -> User | None: ...
    @abstractmethod
    async def save(self, user: User) -> None: ...
    @abstractmethod
    async def delete(self, user_id: str) -> None: ...

class OrderRepository(ABC):
    @abstractmethod
    async def get_by_id(self, order_id: str) -> Order | None: ...
    @abstractmethod
    async def save(self, order: Order) -> None: ...

存储库实现

# infrastructure/repositories/user_repo.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.repositories import UserRepository
from domain.models.user import User, UserRole

class SqlAlchemyUserRepository(UserRepository):
    def __init__(self, session: AsyncSession):
        self._session = session

    async def get_by_id(self, user_id: str) -> User | None:
        model = await self._session.get(UserModel, user_id)
        return self._to_domain(model) if model else None

    async def get_by_email(self, email: str) -> User | None:
        result = await self._session.execute(
            select(UserModel).where(UserModel.email == email)
        )
        model = result.scalar_one_or_none()
        return self._to_domain(model) if model else None

    async def save(self, user: User) -> None:
        model = await self._session.get(UserModel, user.id)
        if model is None:
            model = UserModel(id=user.id)
            self._session.add(model)
        model.email = user.email
        model.name = user.name
        model.role = user.role.value
        model.is_active = user.is_active
        model.created_at = user.created_at

    async def delete(self, user_id: str) -> None:
        model = await self._session.get(UserModel, user_id)
        if model:
            await self._session.delete(model)

    @staticmethod
    def _to_domain(model: UserModel) -> User:
        return User(
            id=model.id, email=model.email, name=model.name,
            role=UserRole(model.role), is_active=model.is_active,
            created_at=model.created_at,
        )

服务层

服务包含业务逻辑。它们依赖存储库抽象，而非具体实现或框架。

# domain/services/user_service.py
from domain.models.user import User, UserRole
from domain.repositories import UserRepository
from domain.errors import NotFoundError, ConflictError

class UserService:
    def __init__(self, user_repo: UserRepository):
        self._user_repo = user_repo

    async def get_user(self, user_id: str) -> User:
        user = await self._user_repo.get_by_id(user_id)
        if user is None:
            raise NotFoundError("User", user_id)
        return user

    async def create_user(self, email: str, name: str) -> User:
        existing = await self._user_repo.get_by_email(email)
        if existing is not None:
            raise ConflictError(f"User with email {email} already exists")

        user = User(
            id=generate_id(),
            email=email,
            name=name,
            role=UserRole.MEMBER,
            created_at=utcnow(),
        )
        await self._user_repo.save(user)
        return user

    async def promote_to_admin(self, user_id: str) -> User:
        user = await self.get_user(user_id)
        user.promote_to_admin()  # Domain logic on the model
        await self._user_repo.save(user)
        return user

    async def deactivate_user(self, user_id: str) -> User:
        user = await self.get_user(user_id)
        user.deactivate()
        await self._user_repo.save(user)
        return user

依赖注入（不使用框架）

使用构造函数注入手动连接依赖。对于大多数 Python 服务，无需 DI 容器。

# api/dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from infrastructure.db import get_session
from infrastructure.repositories.user_repo import SqlAlchemyUserRepository
from domain.services.user_service import UserService

async def get_user_repo(
    session: AsyncSession = Depends(get_session),
) -> SqlAlchemyUserRepository:
    return SqlAlchemyUserRepository(session)

async def get_user_service(
    user_repo: SqlAlchemyUserRepository = Depends(get_user_repo),
) -> UserService:
    return UserService(user_repo)

# api/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from domain.services.user_service import UserService
from domain.errors import NotFoundError, ConflictError
from api.dependencies import get_user_service

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}")
async def get_user(
    user_id: str,
    service: UserService = Depends(get_user_service),
):
    try:
        return await service.get_user(user_id)
    except NotFoundError:
        raise HTTPException(status_code=404, detail="User not found")

错误层级

定义与框架无关的领域错误。在 API 层将其映射为 HTTP 错误。

# domain/errors.py
class DomainError(Exception):
    """Base class for all domain errors."""
    def __init__(self, message: str):
        self.message = message
        super().__init__(message)

class NotFoundError(DomainError):
    def __init__(self, resource: str, resource_id: str | int):
        self.resource = resource
        self.resource_id = resource_id
        super().__init__(f"{resource} {resource_id} not found")

class ConflictError(DomainError):
    pass

class ValidationError(DomainError):
    def __init__(self, field: str, message: str):
        self.field = field
        super().__init__(f"Validation error on {field}: {message}")

class InactiveUserError(DomainError):
    def __init__(self, user_id: str):
        super().__init__(f"User {user_id} is inactive")

class OrderNotEditableError(DomainError):
    def __init__(self, order_id: str, status: str):
        super().__init__(f"Order {order_id} cannot be edited in status '{status}'")

class EmptyOrderError(DomainError):
    def __init__(self, order_id: str):
        super().__init__(f"Order {order_id} cannot be submitted with no items")

将领域错误映射为 HTTP 错误

# api/error_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from domain.errors import DomainError, NotFoundError, ConflictError, ValidationError

def register_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(NotFoundError)
    async def not_found_handler(request: Request, exc: NotFoundError):
        return JSONResponse(status_code=404, content={"error": exc.message})

    @app.exception_handler(ConflictError)
    async def conflict_handler(request: Request, exc: ConflictError):
        return JSONResponse(status_code=409, content={"error": exc.message})

    @app.exception_handler(ValidationError)
    async def validation_handler(request: Request, exc: ValidationError):
        return JSONResponse(
            status_code=422,
            content={"error": exc.message, "field": exc.field},
        )

    @app.exception_handler(DomainError)
    async def domain_error_handler(request: Request, exc: DomainError):
        return JSONResponse(status_code=400, content={"error": exc.message})

测试策略

使用模拟存储库对服务进行单元测试

# tests/unit/test_user_service.py
import pytest
from unittest.mock import AsyncMock
from domain.services.user_service import UserService
from domain.models.user import User, UserRole
from domain.errors import NotFoundError, ConflictError

@pytest.fixture
def mock_repo():
    return AsyncMock()

@pytest.fixture
def service(mock_repo):
    return UserService(user_repo=mock_repo)

@pytest.mark.asyncio
async def test_create_user_success(service, mock_repo):
    mock_repo.get_by_email.return_value = None  # No existing user

    user = await service.create_user(email="[email protected]", name="New User")

    assert user.email == "[email protected]"
    assert user.role == UserRole.MEMBER
    mock_repo.save.assert_called_once()

@pytest.mark.asyncio
async def test_create_user_duplicate_email(service, mock_repo):
    mock_repo.get_by_email.return_value = User(
        id="existing", email="[email protected]", name="Existing",
        role=UserRole.MEMBER, created_at=utcnow(),
    )

    with pytest.raises(ConflictError, match="already exists"):
        await service.create_user(email="[email protected]", name="New")

@pytest.mark.asyncio
async def test_promote_inactive_user_fails(service, mock_repo):
    inactive_user = User(
        id="u1", email="[email protected]", name="X",
        role=UserRole.MEMBER, created_at=utcnow(), is_active=False,
    )
    mock_repo.get_by_id.return_value = inactive_user

    with pytest.raises(InactiveUserError):
        await service.promote_to_admin("u1")

使用测试数据库对存储库进行集成测试

使用内存 SQLite 数据库。在 fixture 中创建表，将会话注入存储库，并测试往返持久化。

# tests/integration/test_user_repo.py
@pytest.fixture
async def session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_sessionmaker(engine)() as session:
        yield session

@pytest.mark.asyncio
async def test_save_and_retrieve(session):
    repo = SqlAlchemyUserRepository(session)
    user = User(id="u1", email="[email protected]", name="Test",
                role=UserRole.MEMBER, created_at=utcnow())
    await repo.save(user)
    await session.commit()

    retrieved = await repo.get_by_id("u1")
    assert retrieved is not None
    assert retrieved.email == "[email protected]"

反模式

在路由处理程序中编写业务逻辑

# BAD: Business logic mixed with HTTP handling
@router.post("/orders")
async def create_order(body: CreateOrderRequest, db = Depends(get_db)):
    user = await db.get(User, body.user_id)
    if not user.is_active:
        raise HTTPException(400, "Inactive user")  # Business rule in handler
    if len(body.items) == 0:
        raise HTTPException(400, "Empty order")    # Business rule in handler
    order = Order(...)
    db.add(order)
    # This is untestable without spinning up FastAPI

# GOOD: Handler delegates to service
@router.post("/orders")
async def create_order(body: CreateOrderRequest, service = Depends(get_order_service)):
    return await service.create_order(body.user_id, body.items)

将 ORM 模型用作领域模型

在业务逻辑中直接使用 SQLAlchemy 模型，会使领域依赖数据库架构。更改列名就会导致业务逻辑出错。

在服务层导入框架

# BAD: Service imports FastAPI
from fastapi import HTTPException

class UserService:
    async def get_user(self, id):
        user = await self.repo.get(id)
        if not user:
            raise HTTPException(404)  # Framework leak!

# GOOD: Service raises domain error
from domain.errors import NotFoundError

class UserService:
    async def get_user(self, id):
        user = await self.repo.get(id)
        if not user:
            raise NotFoundError("User", id)  # Framework-agnostic

没有错误层级

到处使用裸 Exception 或 ValueError，会导致无法将领域错误一致地映射为 HTTP 状态码。应定义以 DomainError 为根的清晰层级。

无法测试的代码

如果不启动 Web 服务器或连接数据库就无法测试服务，那么你的架构就有问题。服务应接受存储库接口，测试则应注入模拟对象。