software-architecture

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Python Backend Architecture

Python 后端架构

Production-ready patterns for Python backend systems following Clean Architecture, Domain-Driven Design, and modern distributed system patterns.

遵循Clean Architecture、Domain-Driven Design及现代分布式系统模式的Python后端系统生产就绪型模式。

Quick Reference

快速参考

Need to...	See Reference
Structure a new project	project-structure.md
Follow code standards	code-style.md
Implement DI, Repository, UoW	architecture-patterns.md
Design entities and value objects	domain-driven-design.md
Build FastAPI endpoints	api-design.md
Set up SQLAlchemy and queries	database-patterns.md
Implement caching	caching.md
Handle events and sagas	event-driven.md
Build microservices	microservices.md
Secure the application	security.md
Add logging, metrics, tracing	observability.md
Write tests	testing.md
Deploy to production	deployment.md

需要...	查看参考文档
搭建新项目结构	project-structure.md
遵循代码标准	code-style.md
实现依赖注入、仓库、工作单元	architecture-patterns.md
设计实体与值对象	domain-driven-design.md
构建FastAPI端点	api-design.md
配置SQLAlchemy与查询	database-patterns.md
实现缓存	caching.md
处理事件与Saga	event-driven.md
构建微服务	microservices.md
保障应用安全	security.md
添加日志、指标、链路追踪	observability.md
编写测试	testing.md
部署到生产环境	deployment.md

Core Principles

核心原则

Architecture Layers

架构分层

┌─────────────────────────────────────────────┐
│           Presentation Layer                │  ← API routes, schemas, middleware
├─────────────────────────────────────────────┤
│           Application Layer                 │  ← Use cases, DTOs, interfaces
├─────────────────────────────────────────────┤
│             Domain Layer                    │  ← Entities, value objects, events
├─────────────────────────────────────────────┤
│          Infrastructure Layer               │  ← DB, cache, messaging, external APIs
└─────────────────────────────────────────────┘

Dependency Rule: Dependencies point inward. Domain has no external dependencies.

┌─────────────────────────────────────────────┐
│           Presentation Layer                │  ← API路由、Schema、中间件
├─────────────────────────────────────────────┤
│           Application Layer                 │  ← 用例、DTO、接口
├─────────────────────────────────────────────┤
│             Domain Layer                    │  ← 实体、值对象、事件
├─────────────────────────────────────────────┤
│          Infrastructure Layer               │  ← 数据库、缓存、消息队列、外部API
└─────────────────────────────────────────────┘

依赖规则：依赖关系向内指向。领域层无外部依赖。

Key Principles

关键原则

Dependency Injection - Depend on abstractions, not implementations
Single Responsibility - One reason to change per class/function
Explicit over Implicit - No hidden dependencies or magic
Async by Default - Use async/await for all I/O operations
Type Hints Everywhere - All public APIs fully typed
Immutability - Prefer immutable data structures (especially value objects)
Early Return - Reduce nesting with guard clauses

依赖注入 - 依赖抽象而非具体实现
单一职责 - 每个类/函数仅一个变更理由
显式优于隐式 - 无隐藏依赖或魔法逻辑
默认异步 - 所有I/O操作使用async/await
全类型提示 - 所有公共API均添加类型注解
不可变性 - 优先使用不可变数据结构（尤其是值对象）
提前返回 - 使用卫语句减少代码嵌套

Architecture Decision Flowchart

架构决策流程图

mermaid

flowchart TD
    A[New Feature] --> B{Complex domain logic?}
    B -->|Yes| C[Use Domain-Driven Design]
    B -->|No| D{Multiple data sources?}

    C --> E[Define Entities & Value Objects]
    E --> F[Create Repository Interfaces]
    F --> G[Implement Use Cases]

    D -->|Yes| H[Use Repository Pattern]
    D -->|No| I{Transaction across aggregates?}

    H --> G

    I -->|Yes| J[Use Unit of Work]
    I -->|No| K[Direct repository calls]

    J --> G
    K --> G

    G --> L{Cross-service operation?}
    L -->|Yes| M[Use Saga Pattern]
    L -->|No| N[Single transaction]

    M --> O[Publish Domain Events]
    N --> O

    O --> P{Needs real-time updates?}
    P -->|Yes| Q[Event-Driven Architecture]
    P -->|No| R[Request-Response]

mermaid

flowchart TD
    A[新功能] --> B{是否存在复杂领域逻辑?}
    B -->|是| C[采用Domain-Driven Design]
    B -->|否| D{是否涉及多数据源?}

    C --> E[定义实体与值对象]
    E --> F[创建仓库接口]
    F --> G[实现用例]

    D -->|是| H[采用仓库模式]
    D -->|否| I{是否跨聚合根事务?}

    H --> G

    I -->|是| J[采用工作单元模式]
    I -->|否| K[直接调用仓库]

    J --> G
    K --> G

    G --> L{是否为跨服务操作?}
    L -->|是| M[采用Saga模式]
    L -->|否| N[单事务]

    M --> O[发布领域事件]
    N --> O

    O --> P[是否需要实时更新?]
    P -->|是| Q[事件驱动架构]
    P -->|否| R[请求-响应模式]

Common Patterns Summary

常见模式总结

Domain Layer

领域层

Entities: Objects with identity and lifecycle. See domain-driven-design.md
Value Objects: Immutable objects compared by value. See domain-driven-design.md
Domain Events: Capture what happened. See domain-driven-design.md
Aggregates: Consistency boundaries. See domain-driven-design.md

实体：具有唯一标识和生命周期的对象。详见domain-driven-design.md
值对象：通过值进行比较的不可变对象。详见domain-driven-design.md
领域事件：捕获领域中发生的事件。详见domain-driven-design.md
聚合根：一致性边界。详见domain-driven-design.md

Application Layer

应用层

Use Cases: Single operation orchestrators
DTOs: Data transfer between layers
Interfaces: Abstract ports for infrastructure

用例：单一操作的编排器
DTO：层间数据传输对象
接口：基础设施层的抽象端口

Infrastructure Layer

基础设施层

Repositories: Data access abstraction. See architecture-patterns.md
Unit of Work: Transaction management. See architecture-patterns.md
Event Bus: Publish/subscribe messaging. See event-driven.md

仓库：数据访问抽象。详见architecture-patterns.md
工作单元：事务管理。详见architecture-patterns.md
事件总线：发布/订阅消息机制。详见event-driven.md

Presentation Layer

表现层

Routers: HTTP endpoint handlers. See api-design.md
Schemas: Request/response validation. See api-design.md
Dependencies: DI factory functions. See api-design.md

路由：HTTP端点处理器。详见api-design.md
Schema：请求/响应校验。详见api-design.md
依赖项：依赖注入工厂函数。详见api-design.md

Error Handling Strategy

错误处理策略

python

undefined

python

undefined

Domain errors (business logic violations)

领域错误（业务逻辑违规）

class DomainError(Exception): pass class EntityNotFoundError(DomainError): pass class BusinessRuleViolationError(DomainError): pass

Infrastructure errors (external system failures)

基础设施错误（外部系统故障）

class InfrastructureError(Exception): pass class DatabaseConnectionError(InfrastructureError): pass class ExternalAPIError(InfrastructureError): pass

Map to HTTP in presentation layer

在表现层映射为HTTP错误

@router.get("/{id}") async def get_item(id: UUID, use_case: GetItemUseCase = Depends(...)): try: return await use_case.execute(id) except EntityNotFoundError: raise HTTPException(status_code=404, detail="Not found") except BusinessRuleViolationError as e: raise HTTPException(status_code=400, detail=str(e)) except InfrastructureError: raise HTTPException(status_code=503, detail="Service unavailable")


See [code-style.md](references/code-style.md#error-handling) for full exception hierarchy.

---

@router.get("/{id}") async def get_item(id: UUID, use_case: GetItemUseCase = Depends(...)): try: return await use_case.execute(id) except EntityNotFoundError: raise HTTPException(status_code=404, detail="未找到") except BusinessRuleViolationError as e: raise HTTPException(status_code=400, detail=str(e)) except InfrastructureError: raise HTTPException(status_code=503, detail="服务不可用")


详见[code-style.md](references/code-style.md#error-handling)获取完整的异常层级结构。

---

Quick Start Patterns

快速入门模式

Basic Repository + Use Case

基础仓库 + 用例

python

undefined

python

undefined

1. Define interface (application layer)

1. 定义接口（应用层）

class UserRepository(ABC): @abstractmethod async def get_by_id(self, id: UUID) -> User | None: pass

@abstractmethod
async def save(self, user: User) -> User: pass

class UserRepository(ABC): @abstractmethod async def get_by_id(self, id: UUID) -> User | None: pass

@abstractmethod
async def save(self, user: User) -> User: pass

2. Implement (infrastructure layer)

2. 实现（基础设施层）

class PostgresUserRepository(UserRepository): def init(self, session: AsyncSession): self._session = session

async def get_by_id(self, id: UUID) -> User | None:
    result = await self._session.execute(
        select(UserModel).where(UserModel.id == id)
    )
    model = result.scalar_one_or_none()
    return self._to_entity(model) if model else None

class PostgresUserRepository(UserRepository): def init(self, session: AsyncSession): self._session = session

async def get_by_id(self, id: UUID) -> User | None:
    result = await self._session.execute(
        select(UserModel).where(UserModel.id == id)
    )
    model = result.scalar_one_or_none()
    return self._to_entity(model) if model else None

3. Use case (application layer)

3. 用例（应用层）

@dataclass class GetUserUseCase: user_repo: UserRepository

async def execute(self, user_id: UUID) -> UserDTO:
    user = await self.user_repo.get_by_id(user_id)
    if not user:
        raise EntityNotFoundError("User", str(user_id))
    return UserDTO.from_entity(user)

@dataclass class GetUserUseCase: user_repo: UserRepository

async def execute(self, user_id: UUID) -> UserDTO:
    user = await self.user_repo.get_by_id(user_id)
    if not user:
        raise EntityNotFoundError("User", str(user_id))
    return UserDTO.from_entity(user)

4. Route (presentation layer)

4. 路由（表现层）

@router.get("/{user_id}") async def get_user( user_id: UUID, use_case: GetUserUseCase = Depends(get_user_use_case) ): return await use_case.execute(user_id)

undefined

@router.get("/{user_id}") async def get_user( user_id: UUID, use_case: GetUserUseCase = Depends(get_user_use_case) ): return await use_case.execute(user_id)

undefined

Adding Caching

添加缓存

python

@dataclass
class GetUserUseCase:
    user_repo: UserRepository
    cache: CacheService

    async def execute(self, user_id: UUID) -> UserDTO:
        # Try cache
        cached = await self.cache.get(f"user:{user_id}")
        if cached:
            return UserDTO.parse_raw(cached)

        # Fetch from DB
        user = await self.user_repo.get_by_id(user_id)
        if not user:
            raise EntityNotFoundError("User", str(user_id))

        # Cache result
        dto = UserDTO.from_entity(user)
        await self.cache.set(f"user:{user_id}", dto.json(), ttl=300)
        return dto

See caching.md for advanced patterns.

python

@dataclass
class GetUserUseCase:
    user_repo: UserRepository
    cache: CacheService

    async def execute(self, user_id: UUID) -> UserDTO:
        # 尝试从缓存获取
        cached = await self.cache.get(f"user:{user_id}")
        if cached:
            return UserDTO.parse_raw(cached)

        # 从数据库获取
        user = await self.user_repo.get_by_id(user_id)
        if not user:
            raise EntityNotFoundError("User", str(user_id))

        # 缓存结果
        dto = UserDTO.from_entity(user)
        await self.cache.set(f"user:{user_id}", dto.json(), ttl=300)
        return dto

详见caching.md获取高级模式。

Publishing Events

发布事件

python

@dataclass
class CreateOrderUseCase:
    order_repo: OrderRepository
    event_bus: EventBus

    async def execute(self, command: CreateOrderCommand) -> OrderDTO:
        order = Order.create(user_id=command.user_id, items=command.items)
        saved = await self.order_repo.add(order)

        # Publish event after successful save
        await self.event_bus.publish(OrderCreatedEvent.from_entity(saved))

        return OrderDTO.from_entity(saved)

See event-driven.md for event handlers and saga patterns.

python

@dataclass
class CreateOrderUseCase:
    order_repo: OrderRepository
    event_bus: EventBus

    async def execute(self, command: CreateOrderCommand) -> OrderDTO:
        order = Order.create(user_id=command.user_id, items=command.items)
        saved = await self.order_repo.add(order)

        # 保存成功后发布事件
        await self.event_bus.publish(OrderCreatedEvent.from_entity(saved))

        return OrderDTO.from_entity(saved)

详见event-driven.md获取事件处理器与Saga模式。

software-architecture

Original

Translation

Python Backend Architecture

Python 后端架构

Quick Reference

快速参考

Core Principles

核心原则

Architecture Layers

架构分层

Key Principles

关键原则

Architecture Decision Flowchart

架构决策流程图

Common Patterns Summary

常见模式总结

Domain Layer

领域层

Application Layer

应用层

Infrastructure Layer

基础设施层

Presentation Layer

表现层

Error Handling Strategy

错误处理策略

Domain errors (business logic violations)

领域错误（业务逻辑违规）

Infrastructure errors (external system failures)

基础设施错误（外部系统故障）

Map to HTTP in presentation layer

在表现层映射为HTTP错误

Quick Start Patterns

快速入门模式

Basic Repository + Use Case

基础仓库 + 用例

1. Define interface (application layer)

1. 定义接口（应用层）

2. Implement (infrastructure layer)

2. 实现（基础设施层）

3. Use case (application layer)

3. 用例（应用层）

4. Route (presentation layer)

4. 路由（表现层）

Adding Caching

添加缓存

Publishing Events

发布事件

Production Readiness

生产就绪检查