fastapi-demo/docs/complete-crud-api/DESIGN_complete-crud-api.md

DESIGN - 完善 CRUD API 功能系统设计

整体架构图

graph TB
    subgraph "API 层"
        A1[GET /api/v1/users/]
        A2[POST /api/v1/users/]
        A3[PUT /api/v1/users/{id}]
        A4[DELETE /api/v1/users/{id}]
        A5[GET /api/v1/products/]
        A6[POST /api/v1/products/]
        A7[PUT /api/v1/products/{id}]
        A8[DELETE /api/v1/products/{id}]
    end

    subgraph "请求验证层"
        B1[UserCreate Schema]
        B2[UserUpdate Schema]
        B3[ProductCreate Schema]
        B4[ProductUpdate Schema]
        B5[Response Schemas]
    end

    subgraph "服务层"
        C1[UserService]
        C2[ProductService]
    end

    subgraph "数据访问层"
        D1[SQLAlchemy Session]
        D2[Database Models]
    end

    subgraph "数据库层"
        E1[(MySQL fast_demo)]
        E2[users table]
        E3[products table]
    end

    A1 --> B5
    A2 --> B1
    A3 --> B2
    A4 --> B5
    A5 --> B5
    A6 --> B3
    A7 --> B4
    A8 --> B5

    B1 --> C1
    B2 --> C1
    B3 --> C2
    B4 --> C2
    B5 --> C1
    B5 --> C2

    C1 --> D1
    C2 --> D1
    D1 --> D2
    D2 --> E1
    E1 --> E2
    E1 --> E3

分层设计

1. API 路由层 (app/api/v1/endpoints/)

职责

• HTTP 请求路由处理
• 请求参数验证
• 响应状态码设置
• 依赖注入管理

核心组件

• users.py: 用户相关 API 端点
• products.py: 产品相关 API 端点

设计原则

• 轻量级路由，业务逻辑下沉到服务层
• 统一的异常处理
• 依赖注入数据库会话

2. 数据验证层 (app/schemas/)

职责

• 请求数据验证
• 响应数据序列化
• 数据类型转换

现有模式

• UserCreate: 创建用户请求数据
• UserUpdate: 更新用户请求数据
• UserResponse: 用户响应数据
• ProductCreate: 创建产品请求数据
• ProductUpdate: 更新产品请求数据
• ProductResponse: 产品响应数据

3. 业务服务层 (app/services/)

职责

• 业务逻辑实现
• 数据库 CRUD 操作
• 业务规则验证
• 异常处理和转换

核心服务类

• UserService: 用户业务逻辑
• ProductService: 产品业务逻辑

设计模式

• 依赖注入数据库会话
• 静态方法实现具体操作
• 统一的异常处理

4. 数据访问层 (app/database/)

职责

• 数据库连接管理
• ORM 模型定义
• 会话生命周期管理

核心组件

• connection.py: 数据库连接和会话管理
• models.py: SQLAlchemy 数据库模型
• base.py: Base 模型类

核心组件设计

UserService 重构设计

class UserService:
    @staticmethod
    def get_all_users(db: Session) -> List[User]:
        """获取所有用户"""

    @staticmethod
    def get_user_by_id(db: Session, user_id: int) -> Optional[User]:
        """根据ID获取用户"""

    @staticmethod
    def create_user(db: Session, user_data: UserCreate) -> User:
        """创建新用户"""

    @staticmethod
    def update_user(db: Session, user_id: int, user_data: UserUpdate) -> Optional[User]:
        """更新用户信息"""

    @staticmethod
    def delete_user(db: Session, user_id: int) -> bool:
        """删除用户"""

ProductService 重构设计

class ProductService:
    @staticmethod
    def get_all_products(db: Session) -> List[Product]:
        """获取所有产品"""

    @staticmethod
    def get_product_by_id(db: Session, product_id: int) -> Optional[Product]:
        """根据ID获取产品"""

    @staticmethod
    def create_product(db: Session, product_data: ProductCreate) -> Product:
        """创建新产品"""

    @staticmethod
    def update_product(db: Session, product_id: int, product_data: ProductUpdate) -> Optional[Product]:
        """更新产品信息"""

    @staticmethod
    def delete_product(db: Session, product_id: int) -> bool:
        """删除产品"""

模块依赖关系图

graph TB
    subgraph "API 端点"
        API1[users.py]
        API2[products.py]
    end

    subgraph "数据模式"
        SCHEMA1[user schemas]
        SCHEMA2[product schemas]
    end

    subgraph "业务服务"
        SERVICE1[UserService]
        SERVICE2[ProductService]
    end

    subgraph "数据库"
        DB1[Database Models]
        DB2[Database Connection]
    end

    API1 --> SCHEMA1
    API1 --> SERVICE1
    API2 --> SCHEMA2
    API2 --> SERVICE2

    SERVICE1 --> DB1
    SERVICE1 --> DB2
    SERVICE2 --> DB1
    SERVICE2 --> DB2

    SCHEMA1 --> DB1
    SCHEMA2 --> DB1

接口契约定义

Users API 接口

POST /api/v1/users/

# Request
{
  "username": "string",
  "email": "string",
  "full_name": "string"
}

# Response (201)
{
  "id": 1,
  "username": "string",
  "email": "string",
  "full_name": "string",
  "is_active": true
}

# Error Response (422)
{
  "detail": "User with email 'test@example.com' already exists"
}

PUT /api/v1/users/{user_id}

# Request (部分更新)
{
  "full_name": "string"  // 可选字段
}

# Response (200)
{
  "id": 1,
  "username": "string",
  "email": "string",
  "full_name": "string",
  "is_active": true
}

# Error Response (404)
{
  "detail": "User not found"
}

DELETE /api/v1/users/{user_id}

# Response (204) - No Content

# Error Response (404)
{
  "detail": "User not found"
}

Products API 接口

POST /api/v1/products/

# Request
{
  "name": "string",
  "description": "string",
  "price": 99.99,
  "stock": 100
}

# Response (201)
{
  "id": 1,
  "name": "string",
  "description": "string",
  "price": 99.99,
  "stock": 100,
  "is_available": true
}

PUT /api/v1/products/{product_id}

# Request (部分更新)
{
  "price": 89.99,
  "stock": 50
}

# Response (200)
{
  "id": 1,
  "name": "string",
  "description": "string",
  "price": 89.99,
  "stock": 50,
  "is_available": true
}

DELETE /api/v1/products/{product_id}

# Response (204) - No Content

# Error Response (404)
{
  "detail": "Product not found"
}

数据流向图

sequenceDiagram
    participant C as Client
    participant API as API Router
    participant V as Validator
    participant S as Service
    participant DB as Database

    Note over C,DB: 创建用户流程
    C->>API: POST /api/v1/users/
    API->>V: 验证 UserCreate
    V->>API: 验证通过
    API->>S: UserService.create_user()
    S->>DB: 检查邮箱唯一性
    DB->>S: 邮箱可用
    S->>DB: 创建用户记录
    DB->>S: 返回创建的用户
    S->>API: 返回 User 对象
    API->>C: 201 + UserResponse

    Note over C,DB: 更新用户流程
    C->>API: PUT /api/v1/users/{id}
    API->>V: 验证 UserUpdate
    V->>API: 验证通过
    API->>S: UserService.update_user()
    S->>DB: 查找用户
    DB->>S: 返回用户或 None
    alt 用户存在
        S->>DB: 更新用户信息
        DB->>S: 返回更新的用户
        S->>API: 返回 User 对象
        API->>C: 200 + UserResponse
    else 用户不存在
        S->>API: 返回 None
        API->>C: 404 Not Found
    end

    Note over C,DB: 删除用户流程
    C->>API: DELETE /api/v1/users/{id}
    API->>S: UserService.delete_user()
    S->>DB: 查找并删除用户
    DB->>S: 返回删除结果
    alt 删除成功
        S->>API: 返回 True
        API->>C: 204 No Content
    else 用户不存在
        S->>API: 返回 False
        API->>C: 404 Not Found
    end

异常处理策略

分层异常处理

1. API 层异常处理

• HTTPException: FastAPI 标准异常
• 状态码映射: 业务异常转换为 HTTP 状态码
• 统一错误格式: 结构化错误响应

2. 服务层异常处理

• 业务异常: 自定义异常类型
• 数据库异常: SQLAlchemy 异常捕获和转换
• 验证异常: Pydantic 验证错误处理

3. 数据库层异常处理

• 连接异常: 数据库连接失败处理
• 约束异常: 唯一性约束、外键约束等
• 事务异常: 事务回滚和重试机制

异常映射表

异常类型	HTTP状态码	错误信息示例
资源不存在	404	"User not found"
唯一性冲突	409	"User with email already exists"
数据验证失败	422	"Invalid email format"
数据库连接异常	500	"Database connection error"
未知异常	500	"Internal server error"

事务管理设计

数据库事务策略

@staticmethod
def create_user(db: Session, user_data: UserCreate) -> User:
    try:
        # 业务逻辑操作
        db_user = User(**user_data.model_dump())
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user
    except IntegrityError as e:
        db.rollback()
        if "email" in str(e):
            raise HTTPException(status_code=409, detail="Email already exists")
        raise HTTPException(status_code=409, detail="Data conflict")
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail="Internal server error")

会话生命周期

• FastAPI 依赖注入: 自动管理会话创建和关闭
• 异常安全: 确保异常时会话正确回滚和关闭
• 连接池: 复用现有连接池配置

性能考虑

数据库查询优化

• 索引使用: 利用现有主键和唯一索引
• 查询优化: 避免 N+1 查询问题
• 批量操作: 必要时支持批量创建和更新

内存管理

• 会话管理: 及时关闭数据库会话
• 对象生命周期: 避免长期持有数据库对象引用

兼容性设计

向后兼容

• 现有 API: 保持现有 GET API 不变
• 响应格式: 保持现有响应结构
• 错误处理: 保持现有错误处理风格

扩展性设计

• 服务接口: 设计可扩展的服务接口
• 模式版本: 支持模式版本演进
• 配置驱动: 通过配置控制功能开关

---

设计完成: ✅ 架构设计完整 创建时间: 2025-09-28 设计师: Claude AI 状态: 准备进入原子化阶段