python-fastapi-ddd-skill

FastAPI + Python DDD 与洋葱架构设计指南

本指南基于 dddpy 参考实现（FastAPI + SQLAlchemy + Python 3.13+），介绍如何使用 DDD 原则和洋葱架构设计 FastAPI 后端。

架构概览

四个同心层，依赖关系指向内部：

Presentation  →  UseCase  →  Infrastructure  →  Domain (innermost)

核心规则：内层永远不依赖外层。Domain 层没有任何外部依赖。

层	职责	示例
Domain	核心业务逻辑，不依赖框架	实体、值对象、Repository 接口、异常
Infrastructure	外部集成	DB Repository、DTO、DI 配置、SQLAlchemy 模型
UseCase	应用工作流	每个用例对应一个带有 execute() 的类
Presentation	HTTP API 接口	FastAPI 路由、Pydantic 模式、错误消息

详细架构指南：请参阅 ARCHITECTURE.md

目录结构

project/
├── main.py
├── app/
│   ├── domain/
│   │   └── {aggregate}/
│   │       ├── entities/
│   │       ├── value_objects/
│   │       ├── repositories/
│   │       └── exceptions/
│   ├── infrastructure/
│   │   ├── di/
│   │   │   └── injection.py
│   │   └── sqlite/
│   │       └── {aggregate}/
│   │           ├── {aggregate}_dto.py
│   │           └── {aggregate}_repository.py
│   ├── usecase/
│   │   └── {aggregate}/
│   │       └── {action}_{aggregate}_usecase.py
│   └── presentation/
│       └── api/
│           └── {aggregate}/
│               ├── handlers/
│               ├── schemas/
│               └── error_messages/
└── tests/

快速参考

1. 实体

实体具有唯一标识符、可变状态和封装的业务逻辑。相等性基于标识，而非属性值。

class Todo:
    def __init__(self, id: TodoId, title: TodoTitle, status: TodoStatus = TodoStatus.NOT_STARTED):
        self._id = id
        self._title = title
        self._status = status

    def __eq__(self, obj: object) -> bool:
        if isinstance(obj, Todo):
            return self.id == obj.id
        return False

    def start(self) -> None:
        self._status = TodoStatus.IN_PROGRESS

    @staticmethod
    def create(title: TodoTitle) -> "Todo":
        return Todo(TodoId.generate(), title)

详细指南：请参阅 ENTITIES.md

2. 值对象

由其值而非标识定义的不可变对象。使用 @dataclass(frozen=True)，并在 __post_init__ 中进行验证。

@dataclass(frozen=True)
class TodoTitle:
    value: str

    def __post_init__(self):
        if not self.value:
            raise ValueError("Title is required")
        if len(self.value) > 100:
            raise ValueError("Title must be 100 characters or less")

详细指南：请参阅 VALUE_OBJECTS.md

3. Repository 接口

在 Domain 层定义抽象接口，并由 Infrastructure 层实现。

class TodoRepository(ABC):
    @abstractmethod
    def save(self, todo: Todo) -> None: ...

    @abstractmethod
    def find_by_id(self, todo_id: TodoId) -> Optional[Todo]: ...

    @abstractmethod
    def find_all(self) -> List[Todo]: ...

    @abstractmethod
    def delete(self, todo_id: TodoId) -> None: ...

详细指南：请参阅 REPOSITORIES.md

4. UseCase

每个用例对应一个类：抽象接口 + 具体实现 + 工厂函数。

class CreateTodoUseCase(ABC):
    @abstractmethod
    def execute(self, title: TodoTitle) -> Todo: ...

class CreateTodoUseCaseImpl(CreateTodoUseCase):
    def __init__(self, todo_repository: TodoRepository):
        self.todo_repository = todo_repository

    def execute(self, title: TodoTitle) -> Todo:
        todo = Todo.create(title=title)
        self.todo_repository.save(todo)
        return todo

def new_create_todo_usecase(repo: TodoRepository) -> CreateTodoUseCase:
    return CreateTodoUseCaseImpl(repo)

详细指南：请参阅 USECASES.md

最佳实践

1. 保持 Domain 层纯粹：Domain 代码中不导入框架（不导入 FastAPI，也不导入 SQLAlchemy）
2. 在层边界使用 DTO：通过 to_entity() / from_entity() 方法在领域实体和基础设施模型之间进行转换
3. 依赖注入：使用 FastAPI 的 Depends() 串联 session → Repository → usecase → handler
4. 一个 UseCase = 一项职责：每个 UseCase 恰好有一个公开的 execute 方法
5. 在值对象中验证：业务规则放在冻结数据类的 __post_init__ 中
6. 领域异常：为违反业务规则的情况创建专用异常类（例如 TodoNotFoundError、TodoAlreadyCompletedError）
7. 工厂函数：公开用于创建实现的 new_* 工厂函数，并将具体类作为实现细节隐藏