Back to Details

python-fastapi-ddd-skill

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI + Python DDD & Onion Architecture Design Guide

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

Guides FastAPI backend design using DDD principles and Onion Architecture, based on the dddpy reference implementation (FastAPI + SQLAlchemy + Python 3.13+).
基于dddpy参考实现（FastAPI + SQLAlchemy + Python 3.13+），指导使用DDD原则和洋葱架构进行FastAPI后端设计。

Architecture Overview

架构概述

Four concentric layers with dependencies pointing inward:
Presentation  →  UseCase  →  Infrastructure  →  Domain (innermost)
Key rule: Inner layers never depend on outer layers. The Domain layer has zero external dependencies.
LayerResponsibilityExamples
DomainCore business logic, no framework depsEntities, Value Objects, Repository interfaces, Exceptions
InfrastructureExternal integrationsDB repos, DTOs, DI config, SQLAlchemy models
UseCaseApplication workflowsOne class per use case with 
execute()
PresentationHTTP API surfaceFastAPI routes, Pydantic schemas, error messages
For detailed architecture guide: See ARCHITECTURE.md
四个同心层，依赖指向内部：
Presentation  →  UseCase  →  Infrastructure  →  Domain (innermost)
核心规则：内层永远不依赖外层，领域层零外部依赖。
层职责示例
领域层（Domain）核心业务逻辑，无框架依赖实体、值对象、仓库接口、异常
基础设施层（Infrastructure）外部集成数据库仓库、DTO、DI配置、SQLAlchemy模型
用例层（UseCase）应用工作流每个用例对应一个带
execute()
的类
展示层（Presentation）HTTP API表层FastAPI路由、Pydantic schemas、错误信息
详细架构指南：参见ARCHITECTURE.md

Directory Structure

目录结构

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/
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/

Quick Reference

快速参考

1. Entity

1. 实体（Entity）

Entities have unique identifiers, mutable state, and encapsulated business logic. Equality is based on identity, not attribute values.
python
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)
Detailed guide: See ENTITIES.md
实体具备唯一标识符、可变状态和封装的业务逻辑，相等性基于标识而非属性值。
python
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. Value Object

2. 值对象（Value Object）

Immutable objects defined by their values, not identity. Use 
@dataclass(frozen=True)
with validation in 
__post_init__
.
python
@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")
Detailed guide: See VALUE_OBJECTS.md
由值而非标识定义的不可变对象，使用
@dataclass(frozen=True)
，在
__post_init__
中实现校验。
python
@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 Interface

3. 仓库接口（Repository Interface）

Define abstract interfaces in the Domain layer. Infrastructure implements them.
python
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: ...
Detailed guide: See REPOSITORIES.md
在领域层定义抽象接口，由基础设施层实现。
python
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

4. 用例（UseCase）

One class per use case. Abstract interface + concrete implementation + factory function.
python
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)
Detailed guide: See USECASES.md
每个用例对应一个类，包含抽象接口+具体实现+工厂函数。
python
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

Best Practices

最佳实践

  1. Keep Domain Layer Pure: No framework imports (no FastAPI, no SQLAlchemy) in domain code
  2. Use DTOs at Layer Boundaries: Convert between domain entities and infrastructure models via 
    to_entity()
    / 
    from_entity()
    methods
  3. Dependency Injection: Use FastAPI's 
    Depends()
    to wire session → repository → usecase → handler
  4. One UseCase = One Responsibility: Each UseCase has exactly one public 
    execute
    method
  5. Validate in Value Objects: Business rules live in 
    __post_init__
    of frozen dataclasses
  6. Domain Exceptions: Create specific exception classes for business rule violations (e.g., 
    TodoNotFoundError
    , 
    TodoAlreadyCompletedError
    )
  7. Factory Functions: Expose 
    new_*
    factory functions for creating implementations, keeping concrete classes as implementation details
  1. 保持领域层纯净：领域代码中不要导入框架（不能引入FastAPI、SQLAlchemy相关内容）
  2. 在层边界使用DTO：通过
    to_entity()
    / 
    from_entity()
    方法实现领域实体和基础设施模型之间的转换
  3. 依赖注入：使用FastAPI的
    Depends()
    串联会话→仓库→用例→处理器
  4. 一个用例对应一个职责：每个用例仅对外暴露一个
    execute
    公共方法
  5. 在值对象中做校验：业务规则存放在冻结数据类的
    __post_init__
    方法中
  6. 领域异常：为业务规则违规场景创建特定的异常类（例如
    TodoNotFoundError
    、
    TodoAlreadyCompletedError
    ）
  7. 工厂函数：暴露
    new_*
    工厂函数创建实现类实例，将具体类作为实现细节隐藏