FastAPI 后端开发规范

触发条件

• Develop FastAPI projects
• Implement async REST APIs
• Use SQLAlchemy 2.0 async ORM
• Use Pydantic v2 for data validation
• Implement JWT / Redis Token authentication
• Implement role-based permission control
• Develop youlai-fastapi admin backend modules

Part 1: 技术栈

Part 2: 目录结构

目录结构以社区标杆 zhanymkanov/fastapi-best-practices 为准。

youlai-fastapi/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 入口：挂载路由、中间件、异常处理器
│   ├── config.py               # Pydantic Settings
│   ├── database.py             # 异步引擎 + session + get_db
│   ├── redis.py                # Redis 连接池 + get_redis
│   ├── response.py             # Result[T] + ResultCode 统一响应
│   ├── exceptions.py           # BusinessException + 全局异常处理器
│   ├── enums.py                 # 全局枚举
│   ├── pagination.py           # PageQuery / PageResult
│   ├── constants.py            # 全局常量：安全前缀 / REDIS_ 前缀 / 系统级
│   ├── dependencies.py         # 全局依赖 get_current_user / require_perm
│   ├── middleware.py           # CORS + 请求日志 + 限流
│   ├── registry.py             # 导入全部域模型，触发 Base.metadata 注册（Base/Mixin 在 database.py）
│   ├── auth/                   # 认证域：router.py / schemas.py / service.py / token.py / utils.py
│   ├── captcha/                # 验证码：service.py / constants.py
│   ├── system/                 # 系统管理域
│   │   ├── user/  role/  menu/  dept/  dict/  log/
│   │   │   └── 各含 router.py  schemas.py  service.py  models.py
│   │   │       （role/ 含 constants.py、data_permission.py；log/ 含 constants.py、operation_log.py）
│   │   ├── config/             # 仅 models.py + router.py（配置内联，无 schemas/service）
│   │   └── notice/             # 仅 models.py + router.py
│   └── tool/                   # 工具与集成域
│       └── file/  codegen/  wxma/  sse/   # 文件 / 代码生成 / 微信小程序 / 服务端推送
├── alembic/                    # 数据库迁移
├── alembic.ini
├── tests/
│   ├── conftest.py
│   ├── auth/test_auth.py       # 按业务域镜像
│   └── test_result_code.py     # 全局
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml              # uv 依赖管理
├── .env.example
├── .gitignore
└── README.md

设计原则：

• 按业务域组织：每个域目录自包含 router/schemas/models/service（按需含 constants/dependencies/exceptions），域间通过显式模块名导入。
• models/ 仅放 ORM 基类与全域注册，纯 ORM 不依赖任何业务层。
• 全局基础设施（config / database / redis / response / exceptions / pagination / constants / dependencies / middleware）平铺在 app/ 根，被各域共享。
• 不设 core/ common/ framework/ modules/ 分层包。

Part 3: 命名规范

3.1 文件命名

3.2 类命名

Schema 后缀规范：

3.3 方法命名

3.4 变量命名

关键约定：ORM 模型字段用 snake_case（对应 PostgreSQL 列名），Schema/API 字段用 camelCase（对应前端 JSON 键名）。

3.5 import 排序

# 1. 标准库
import io
import logging

# 2. 第三方库
from fastapi import APIRouter, Depends
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, Field

# 3. 项目内部（按层排序）
from app.database import get_db
from app.constants import ROOT_ROLE_CODE
from app.pagination import PageResult
from app.dependencies import require_perm
from app.exceptions import BusinessException
from app.response import Result, ResultCode
from app.user.models import SysUser
from app.user.schemas import UserCreate, UserVO

Part 4: 路由与 API 规范

4.1 路由前缀

router = APIRouter(prefix="/api/v1/users", tags=["用户管理"])

4.2 标准 CRUD 路径

4.3 路由注册顺序（重要！）

固定路径端点必须定义在路径参数端点之前，否则 /{user_id} 会先被匹配：

# ✅ 正确顺序
@router.get("")                    # GET /api/v1/users
@router.get("/template")           # 固定路径 —— 必须在 /{user_id} 前
@router.get("/export")             # 固定路径
@router.post("/import")            # 固定路径
@router.patch("/status")           # 固定路径
@router.patch("/password")         # 固定路径

@router.post("")                   # POST /api/v1/users
@router.get("/{user_id}")          # 路径参数 —— 最后注册
@router.put("/{user_id}")          # 路径参数
@router.delete("/{ids}")           # 路径参数

4.4 Router 函数模板

# router.py — 完整模板

import io
import logging
from fastapi import APIRouter, Depends, Query, UploadFile, File
from sqlalchemy.ext.asyncio import AsyncSession
from openpyxl import Workbook

from app.database import get_db
from app.dependencies import require_perm
from app.response import Result
from app.user.schemas import UserCreate, UserQuery, UserUpdate, UserVO
from app.user.service import UserService

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/users", tags=["用户管理"])


@router.get("", summary="用户分页列表",
    dependencies=[Depends(require_perm("sys:user:list"))])
async def get_user_page(
    pageNum: int = Query(default=1, ge=1),
    pageSize: int = Query(default=10, ge=1, le=100),
    keywords: str | None = Query(default=None),
    status: int | None = Query(default=None),
    db: AsyncSession = Depends(get_db),
):
    """分页查询用户列表。

    - **pageNum**: 当前页码
    - **pageSize**: 每页条数（1-100）
    - **keywords**: 搜索关键词（用户名/昵称/手机号）
    - **status**: 状态筛选
    """
    query = UserQuery(pageNum=pageNum, pageSize=pageSize,
                      keywords=keywords, status=status)
    result = await UserService(db).get_page(query)
    return Result(data=result)


@router.get("/{user_id}", summary="用户详情",
    dependencies=[Depends(require_perm("sys:user:detail"))])
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    """根据 ID 获取用户详情。"""
    vo = await UserService(db).get_by_id(user_id)
    return Result(data=vo)


@router.post("", summary="创建用户",
    dependencies=[Depends(require_perm("sys:user:create"))])
async def create_user(form: UserCreate, db: AsyncSession = Depends(get_db)):
    """创建新用户，自动分配角色。"""
    vo = await UserService(db).create(form)
    return Result(data=vo)


@router.put("/{user_id}", summary="更新用户",
    dependencies=[Depends(require_perm("sys:user:edit"))])
async def update_user(user_id: int, form: UserUpdate,
                      db: AsyncSession = Depends(get_db)):
    """更新用户信息及角色。"""
    form.id = user_id
    vo = await UserService(db).update(form)
    return Result(data=vo)


@router.delete("/{ids}", summary="删除用户",
    dependencies=[Depends(require_perm("sys:user:delete"))])
async def delete_users(ids: str, db: AsyncSession = Depends(get_db)):
    """批量逻辑删除用户。"""
    count = await UserService(db).delete(ids)
    return Result(data=count, msg=f"成功删除 {count} 条记录")

Part 5: HTTP 状态码与业务错误码

5.1 设计理念：双轨制（HTTP 语义 + 业务码）

背景：社区存在两派争论——"统一返回 200，用 body 的 code 区分" vs "用正确的 HTTP 状态码"。本规范采用 RFC 9457 Problem Details 思路 + 兼容 youlai-boot/vue3-element-admin 前端拦截器的双轨制方案。

双轨制原则：

HTTP 状态码 → 大类划分（2xx 成功 / 4xx 客户端错误 / 5xx 服务端错误）
业务错误码  → 细粒度分类（A0230 Token 过期 / A0402 密码错误 / B0003 数据不存在）

• HTTP 状态码：协议层，用于网关/监控/API 工具识别请求成败
• 业务错误码：应用层，用于前端精确判断业务逻辑
• 两者共存：每个错误响应同时携带正确的 HTTP 状态码和 body 中的 code 字段

5.2 HTTP 状态码使用规范

为什么不用 4xx 表示所有业务错误？
HTTP 状态码应反映协议层的成败。"密码错误"是业务校验失败而非请求格式错误，用 400 会与参数格式错误混淆。本项目将"密码错误"归为 401（认证失败），"用户名不存在"归为 404（资源不存在），"唯一键冲突"归为 409（冲突）——选择语义最匹配的状态码。

5.3 业务错误码（ResultCode）

from enum import Enum


class ResultCode(str, Enum):
    """业务错误码 — 与 HTTP 状态码配合使用。

    首位编码：
      0 - 成功
      A - 用户端错误（4xx）
      B - 系统端错误（5xx）
      C - 第三方服务错误（5xx）
    """

    # ── 成功 ──
    SUCCESS = "00000"                    # → HTTP 200

    # ── 认证/授权类（4xx）──
    TOKEN_INVALID = "A0230"              # → HTTP 401 访问令牌无效或过期
    TOKEN_REFRESH_FAIL = "A0231"         # → HTTP 401 刷新令牌无效或过期
    ACCESS_DENIED = "A0301"              # → HTTP 403 权限不足

    # ── 参数/校验类（4xx）──
    PARAM_VALID_FAIL = "A0400"           # → HTTP 422 参数校验失败
    USERNAME_NOT_FOUND = "A0401"         # → HTTP 404 用户名不存在
    BAD_CREDENTIALS = "A0402"            # → HTTP 401 密码错误
    USER_DISABLED = "A0403"             # → HTTP 403 用户被禁用
    CAPTCHA_ERROR = "A0404"             # → HTTP 400 验证码错误

    # ── 业务异常 ──
    SYSTEM_ERROR = "B0001"              # → HTTP 500 系统执行异常（全局兜底）
    DUPLICATE_KEY = "B0002"              # → HTTP 409 数据重复
    DATA_NOT_FOUND = "B0003"             # → HTTP 404 数据不存在
    OPERATE_DENIED = "B0004"             # → HTTP 403 操作不允许

    @property
    def http_status(self) -> int:
        """业务错误码 → 对应的 HTTP 状态码。"""
        mapping = {
            "00000": 200,
            "A0230": 401, "A0231": 401,
            "A0301": 403, "A0403": 403, "B0004": 403,
            "A0400": 422,
            "A0401": 404, "B0003": 404,
            "A0402": 401,
            "A0404": 400,
            "B0002": 409,
            "B0001": 500,
        }
        return mapping.get(self.value, 500)

5.4 BusinessException 与全局异常处理器

# app/exceptions.py

from fastapi import Request
from fastapi.responses import JSONResponse
from app.response import ResultCode


class BusinessException(Exception):
    """业务异常 — 携带业务错误码和 HTTP 状态码。"""

    def __init__(self, code: ResultCode = ResultCode.SYSTEM_ERROR, msg: str = "系统异常"):
        self.code = code
        self.msg = msg
        self.http_status = code.http_status  # ← 自动映射 HTTP 状态码
        super().__init__(msg)


async def business_exception_handler(request: Request, exc: BusinessException) -> JSONResponse:
    """业务异常处理 — 返回语义化 HTTP 状态码 + body 中的 code。"""
    return JSONResponse(
        status_code=exc.http_status,            # ← HTTP 401/403/404/409/500...
        content={"code": exc.code.value, "msg": exc.msg, "data": None},
    )


async def validation_exception_handler(request: Request, exc) -> JSONResponse:
    """参数校验异常 — HTTP 422 + body code A0400。"""
    return JSONResponse(
        status_code=422,
        content={"code": ResultCode.PARAM_VALID_FAIL.value, "msg": str(exc), "data": None},
    )


async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """全局兜底 — HTTP 500 + body code B0001。"""
    return JSONResponse(
        status_code=500,
        content={"code": ResultCode.SYSTEM_ERROR.value, "msg": "系统执行异常", "data": None},
    )

5.5 业务异常抛出示例

# ✅ 认证失败 → HTTP 401
raise BusinessException(code=ResultCode.TOKEN_INVALID, msg="访问令牌无效或过期")
raise BusinessException(code=ResultCode.BAD_CREDENTIALS, msg="密码错误")

# ✅ 权限不足 → HTTP 403
raise BusinessException(code=ResultCode.ACCESS_DENIED, msg="权限不足")

# ✅ 数据不存在 → HTTP 404
raise BusinessException(code=ResultCode.DATA_NOT_FOUND, msg="用户不存在")
raise BusinessException(code=ResultCode.USERNAME_NOT_FOUND, msg="用户名不存在")

# ✅ 唯一键冲突 → HTTP 409
raise BusinessException(code=ResultCode.DUPLICATE_KEY, msg="用户名已存在")

# ✅ 操作不允许 → HTTP 403
raise BusinessException(code=ResultCode.OPERATE_DENIED, msg="存在子菜单，无法删除")

# ✅ 系统异常 → HTTP 500（由全局异常处理器兜底，通常不手动抛出）

5.6 与前端 axios 拦截器的兼容性

vue3-element-admin 的 axios 拦截器同时处理两个分支，天然兼容双轨制：

// vue3-element-admin/src/utils/request.ts（已验证）
http.interceptors.response.use(
  (response) => {
    // HTTP 2xx 进入此分支
    const { code, data, msg } = response.data;
    if (code === "00000") return data;        // 业务成功
    ElMessage.error(msg);                      // 业务错误（HTTP 200 但 code 非 00000）
    return Promise.reject(new Error(msg));
  },
  (error) => {
    // HTTP 4xx/5xx 进入此分支
    const { code, msg } = error.response.data;
    if (code === "A0230") { /* Token 过期 → 自动刷新 */ }
    if (code === "A0231") { /* Refresh 失败 → 跳转登录 */ }
    if (code === "A0301") { /* 权限不足 → 刷新权限 */ }
    ElMessage.error(msg);
    return Promise.reject(error);
  }
);

关键：前端无论进入哪个分支，都从 response.data.code 读取业务码。双轨制方案前端无需任何改动。

5.7 错误码设计规范

第二位子分类（A 类）：

新增错误码流程：

1. 在 ResultCode 枚举中添加新值
2. 在 http_status property 的 mapping 中添加映射
3. 更新本规范的 5.3 节表格

Part 6: 统一响应格式

6.1 Result 结构

{
    "code": "00000",
    "msg": "成功",
    "data": { … }
}

from pydantic import BaseModel, Field
from typing import Generic, TypeVar

T = TypeVar("T")

class Result(BaseModel, Generic[T]):
    code: str = Field(default="00000", description="业务状态码")
    msg: str = Field(default="成功", description="提示信息")
    data: T | None = Field(default=None, description="响应数据")

6.2 PageResult 结构

{
    "code": "00000",
    "msg": "成功",
    "data": {
        "list": [ … ],
        "total": 100,
        "pageNum": 1,
        "pageSize": 10
    }
}

class PageResult(BaseModel):
    list: list = Field(default_factory=list)
    total: int = Field(default=0)
    pageNum: int = Field(default=1)
    pageSize: int = Field(default=10)

6.3 分页查询参数

class PageQuery(BaseModel):
    pageNum: int = Field(default=1, ge=1)
    pageSize: int = Field(default=10, ge=1, le=100)

前端约定：分页参数命名为 pageNum / pageSize（camelCase），兼容 vue3-element-admin 的 TablePage 组件。

6.4 返回示例

# 成功返回
return Result(data=user_vo)

# 成功返回（自定义消息）
return Result(data=count, msg=f"成功删除 {count} 条记录")

# 成功返回（无数据）
return Result(data=None)

# 错误返回
return Result(code=ResultCode.CAPTCHA_ERROR, msg="验证码错误", data=None)

Part 7: ORM 模型规范

7.1 基类 Mixin

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import BigInteger, DateTime, Integer, func


class Base(DeclarativeBase):
    """ORM 声明式基类。"""
    pass


class TimestampMixin:
    """自动管理 create_time / update_time。"""
    create_time: Mapped[datetime | None] = mapped_column(
        DateTime, server_default=func.now(), comment="创建时间"
    )
    update_time: Mapped[datetime | None] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间"
    )


class SoftDeleteMixin:
    """逻辑删除标识。"""
    is_deleted: Mapped[int] = mapped_column(
        Integer, default=0, server_default="0", comment="逻辑删除 0-未删除 1-已删除"
    )


class BaseIdMixin:
    """自增主键 ID。"""
    id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True, comment="主键ID"
    )

7.2 ORM 模型示例

class SysUser(Base, BaseIdMixin, TimestampMixin, SoftDeleteMixin):
    __tablename__ = "sys_user"

    username: Mapped[str] = mapped_column(String(64), unique=True, comment="用户名")
    nickname: Mapped[str] = mapped_column(String(64), comment="昵称")
    password: Mapped[str] = mapped_column(String(100), comment="密码")
    dept_id: Mapped[int | None] = mapped_column(BigInteger, comment="部门ID")
    mobile: Mapped[str | None] = mapped_column(String(20), comment="手机号")
    email: Mapped[str | None] = mapped_column(String(100), comment="邮箱")
    status: Mapped[int] = mapped_column(Integer, default=1, server_default="1",
                                         comment="状态 1-启用 0-禁用")

7.3 字段命名对照

转换规则：ORM → VO 时，tovo 或 tovo 方法负责 snakecase → camelCase 转换。

7.4 VO 转换模板

@staticmethod
def _to_vo(m: SysUser) -> UserVO:
    """ORM 模型 → 视图对象（snake_case → camelCase）。"""
    return UserVO(
        id=m.id,
        username=m.username,
        deptId=m.dept_id,       # ← 手动映射
        createTime=str(m.create_time) if m.create_time else None,
        # ...
    )

Part 8: 认证与鉴权

8.1 Token 模式

通过 SESSION_TYPE 环境变量切换：

8.2 获取当前用户

from app.dependencies import get_current_user

@router.get("/me")
async def my_info(user: SysUserDetails = Depends(get_current_user)):
    return Result(data={"userId": user.userId, "username": user.username})

8.3 权限校验

from app.dependencies import require_perm

# 精确权限
@router.post("", dependencies=[Depends(require_perm("sys:user:create"))])
async def create_user(form: UserCreate, db: AsyncSession = Depends(get_db)):
    ...

# 登录即可（不检查权限）
@router.get("/options", dependencies=[Depends(require_perm())])
async def get_options(db: AsyncSession = Depends(get_db)):
    ...

8.4 权限标识命名规范

模块:实体:操作
sys:user:create
sys:user:edit
sys:user:delete
sys:role:list
sys:menu:list
sys:dept:list

Part 9: 数据库操作规范

9.1 异步查询

# ✅ 正确：使用 async/await
result = await self.db.execute(
    select(SysUser).where(SysUser.id == user_id, SysUser.is_deleted == 0)
)
user = result.scalar_one_or_none()

# ✅ 正确：分页查询
rows = await self.db.execute(
    select(SysUser)
    .where(*conditions)
    .order_by(SysUser.create_time.desc())
    .offset(offset)
    .limit(page_size)
)
users = rows.scalars().all()

# ✅ 正确：原生 SQL（复杂联表查询）
result = await self.db.execute(
    text("""
        SELECT r.code FROM sys_role r
        INNER JOIN sys_user_role ur ON r.id = ur.role_id
        WHERE ur.user_id = :user_id AND r.is_deleted = 0 AND r.status = 1
    """),
    {"user_id": user_id},
)

9.2 事务管理

# 由 get_db() 依赖自动管理：
# yield 前 → session 开始
# yield 后 → commit（成功）或 rollback（异常）
# finally → session.close()

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()     # 自动提交
        except Exception:
            await session.rollback()   # 自动回滚
            raise
        finally:
            await session.close()

9.3 批量操作

# ✅ 批量删除（原生 SQL）
await self.db.execute(
    text("UPDATE sys_user SET is_deleted = 1 WHERE id = ANY(:ids)"),
    {"ids": id_list},
)

# ✅ 批量插入
for data in batch:
    self.db.add(SysConfig(**data))
await self.db.flush()

Part 10: 添加新模块 — 完整教程

以添加 "数据字典项" 模块为例，展示从零到完整的步骤。

Step 1: 检查 ORM 模型

确认 app/system/dict/models.py 中已有对应的 ORM 模型。若无，创建：

# app/system/dict/models.py
class SysDictItem(Base, BaseIdMixin):
    __tablename__ = "sys_dict_item"
    dict_code: Mapped[str | None] = mapped_column(String(50))
    value: Mapped[str | None] = mapped_column(String(50))
    label: Mapped[str | None] = mapped_column(String(100))
    tag_type: Mapped[str | None] = mapped_column(String(50))
    status: Mapped[int] = mapped_column(Integer, default=1)
    sort: Mapped[int] = mapped_column(Integer, default=0)

Step 2: 创建模块目录

mkdir app/dict_item

Step 3: 创建 schemas.py

# app/dict_item/schemas.py
"""字典项 Schemas。"""
from pydantic import BaseModel, Field


class DictItemCreate(BaseModel):
    """创建字典项表单。"""
    dictCode: str = Field(..., max_length=50, description="关联字典编码")
    value: str = Field(..., max_length=50, description="字典项值")
    label: str = Field(..., max_length=100, description="字典项标签")
    tagType: str | None = Field(default=None, max_length=50, description="标签类型")
    status: int = Field(default=1, description="状态 1-正常 0-禁用")
    sort: int = Field(default=0, description="排序")


class DictItemUpdate(DictItemCreate):
    """更新字典项表单。"""
    id: int = Field(..., description="字典项ID")


class DictItemVO(BaseModel):
    """字典项视图对象。"""
    id: int | None = None
    dictCode: str = ""
    value: str = ""
    label: str = ""
    tagType: str | None = None
    status: int = 1
    sort: int = 0
    model_config = {"from_attributes": True}


class DictItemQuery(BaseModel):
    """字典项查询参数。"""
    dictCode: str = Field(..., description="字典编码")

Step 4: 创建 service.py

# app/dict_item/service.py
"""字典项服务。"""
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.exceptions import BusinessException
from app.response import ResultCode
from app.dict.models import SysDictItem
from app.dict_item.schemas import (
    DictItemCreate, DictItemUpdate, DictItemVO, DictItemQuery,
)


class DictItemService:
    """字典项管理服务。"""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_items(self, dict_code: str) -> list[DictItemVO]:
        """根据字典编码查询字典项列表。"""
        rows = await self.db.execute(
            select(SysDictItem)
            .where(SysDictItem.dict_code == dict_code)
            .order_by(SysDictItem.sort.asc())
        )
        items = rows.scalars().all()
        return [DictItemVO.model_validate(it, from_attributes=True) for it in items]

    async def get_by_id(self, item_id: int) -> DictItemVO:
        """根据 ID 查询字典项。"""
        obj = await self.db.get(SysDictItem, item_id)
        if obj is None:
            raise BusinessException(code=ResultCode.DATA_NOT_FOUND, msg="字典项不存在")
        return DictItemVO.model_validate(obj, from_attributes=True)

    async def create(self, form: DictItemCreate) -> DictItemVO:
        """创建字典项。"""
        obj = SysDictItem(**form.model_dump())
        self.db.add(obj)
        await self.db.flush()
        return DictItemVO.model_validate(obj, from_attributes=True)

    async def update(self, form: DictItemUpdate) -> DictItemVO:
        """更新字典项。"""
        obj = await self.db.get(SysDictItem, form.id)
        if obj is None:
            raise BusinessException(code=ResultCode.DATA_NOT_FOUND, msg="字典项不存在")
        for key, val in form.model_dump(exclude={"id"}).items():
            setattr(obj, key, val)
        await self.db.flush()
        return DictItemVO.model_validate(obj, from_attributes=True)

    async def delete(self, ids: str) -> int:
        """批量删除字典项。"""
        id_list = [int(x) for x in ids.split(",") if x.strip()]
        for iid in id_list:
            obj = await self.db.get(SysDictItem, iid)
            if obj:
                await self.db.delete(obj)
        await self.db.flush()
        return len(id_list)

Step 5: 创建 router.py

# app/dict_item/router.py
"""字典项路由。"""
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.dependencies import require_perm
from app.response import Result
from app.dict_item.schemas import (
    DictItemCreate, DictItemUpdate,
)
from app.dict_item.service import DictItemService

router = APIRouter(prefix="/api/v1/dict/items", tags=["字典管理"])


@router.get("/{dict_code}", summary="字典项列表")
async def get_dict_items(dict_code: str, db: AsyncSession = Depends(get_db)):
    items = await DictItemService(db).get_items(dict_code)
    return Result(data=items)


@router.get("/{item_id}/detail", summary="字典项详情")
async def get_dict_item(item_id: int, db: AsyncSession = Depends(get_db)):
    vo = await DictItemService(db).get_by_id(item_id)
    return Result(data=vo)


@router.post("", summary="创建字典项",
    dependencies=[Depends(require_perm("sys:dict:create"))])
async def create_dict_item(form: DictItemCreate,
    db: AsyncSession = Depends(get_db)):
    vo = await DictItemService(db).create(form)
    return Result(data=vo)


@router.put("/{item_id}", summary="更新字典项",
    dependencies=[Depends(require_perm("sys:dict:edit"))])
async def update_dict_item(item_id: int, form: DictItemUpdate,
    db: AsyncSession = Depends(get_db)):
    form.id = item_id
    vo = await DictItemService(db).update(form)
    return Result(data=vo)


@router.delete("/{ids}", summary="删除字典项",
    dependencies=[Depends(require_perm("sys:dict:delete"))])
async def delete_dict_items(ids: str, db: AsyncSession = Depends(get_db)):
    count = await DictItemService(db).delete(ids)
    return Result(data=count)

Step 6: 创建 init.py

# app/dict_item/__init__.py

Step 7: 在 main.py 注册路由

# app/main.py — create_app() 函数中追加
from app.dict_item.router import router as dict_item_router
app.include_router(dict_item_router)

Step 8: 验证

# 启动服务
fastapi dev app/main.py

# 访问 Swagger 确认路由已注册
open http://localhost:8000/api/v1/swagger-ui.html

模块文件清单

app/dict_item/
├── __init__.py
├── schemas.py       # ← 第一步创建
├── service.py       # ← 第二步创建
└── router.py        # ← 第三步创建

扩展：单文件模块（简单 CRUD）

对于日志管理等简单模块，可将 schemas + service + router 合并为一个文件：

# app/system/log/router.py
# ┌── schemas ──┐
class LogQuery(BaseModel): ...
class LogVO(BaseModel): ...

# ┌── service ──┐
class LogService: ...
    async def get_page(self, query): ...

# ┌── router ──┐
router = APIRouter(prefix="/api/v1/logs", tags=["日志管理"])

@router.get("", ...)
async def get_logs(...): ...

选择标准：单个实体 < 5 个接口 → 单文件模块；> 5 个接口或有复杂查询 → 拆分为 schemas + service + router。

Part 11: 注释规范

来源：PEP 257 (Docstring Conventions) + Google Style Docstrings。

11.1 总则

• docstring 优先：所有公共模块、类、函数必须有 docstring
• 自解释代码：好的命名和结构是自解释的，注释描述"为什么"而非"如何做"
• 中文优先：与其用半吊子英文，不如用中文把问题说清楚；专有名词保留英文（如 HTTP、async）
• 同步更新：代码修改时 docstring 必须同步修改

11.2 docstring 格式（Google 风格）

async def get_user_page(
    pageNum: int = Query(default=1, ge=1),
    pageSize: int = Query(default=10, ge=1, le=100),
    keywords: str | None = Query(default=None),
    db: AsyncSession = Depends(get_db),
) -> PageResult:
    """分页查询用户列表。

    根据关键词模糊搜索用户名/昵称/手机号，支持状态筛选。

    Args:
        pageNum: 当前页码，从 1 开始。
        pageSize: 每页条数，1-100。
        keywords: 搜索关键词（用户名/昵称/手机号），为空时查全部。
        db: 异步数据库会话。

    Returns:
        分页结果，包含 list 和 total。

    Raises:
        BusinessException: 当 pageSize 超过 100 时（通常由 Pydantic 校验拦截）。
    """
    ...

11.3 各层注释要点

11.4 常见反模式

11.5 TODO / FIXME 标记规范

Part 12: 代码质量检查清单

11.1 通用检查

• 所有公共类、方法有 docstring
• 用 Pydantic v2 的 model_dump() 而非 dict()
• 用 modelvalidate(obj, fromattributes=True) 做 ORM → Schema 转换
• 所有 .py 文件有 # -- coding: utf-8 -- 或 UTF-8 编码
• import 按标准库 → 第三方 → 内部顺序排列
• 无未使用的 import
• 无 print() 调试代码，统一用 logger
• 日志不输出密码、密钥等敏感信息
• 配置通过 Settings + .env 管理，不硬编码
• 所有 async def 函数内部使用 await

11.2 数据库相关

• ORM 模型字段名使用 snake_case，与数据库列名一致
• 所有查询加上 is_deleted == 0 过滤（如模型有 SoftDeleteMixin）
• 使用 select 而非 Model.query（SQLAlchemy 2.0+ 风格）
• 大量数据操作使用 flush() 而非频繁 commit()
• ScalarResult.scalaroneor_none() 而非 first()（区分 None vs 第一条）

11.3 API 相关

• 路由 prefix 统一 "/api/v1/模块名"
• 固定路径端点定义在路径参数端点之前
• 分页参数命名为 pageNum / pageSize
• JSON 字段使用 camelCase（deptId、createTime）
• 统一返回 Result[T] 包装（body 含 code/msg/data）
• HTTP 状态码语义化：成功 200 / 认证失败 401 / 权限不足 403 / 不存在 404 / 冲突 409 / 校验 422 / 系统错误 500
• 业务错误码与 HTTP 状态码双轨配合（见 Part 5）
• ResultCode.http_status property 映射正确
• 错误返回使用 ResultCode 枚举，不硬编码字符串
• 敏感接口添加 dependencies=[Depends(require_perm(...))]
• tree_path 在 add + flush 之后再设置（拿到自增 id）

11.4 Service 层

• Service 构造函数接受 db: AsyncSession
• 数据不存在时抛出 BusinessException(code=ResultCode.DATANOTFOUND)
• 唯一键冲突时抛出 BusinessException(code=ResultCode.DUPLICATE_KEY)
• 操作不允许时抛出 BusinessException(code=ResultCode.OPERATE_DENIED)
• 批量操作前检查参数非空
• tovo / totree 等私有方法前缀 _

Part 13: 常见反模式

Part 14: 单模块文件

对于 config、notice、log 等简单模块，可将 schemas + service + router 合并到单个 router.py 文件中，减少文件碎片化。

Part 15: Docker 部署

# docker/docker-compose.yml
services:
  api:
    build: ..
    ports: ["8000:8000"]
    depends_on: [postgres, redis]
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:123456@postgres:5432/youlai_admin
      REDIS_URL: redis://redis:6379/0
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: youlai_admin
      POSTGRES_PASSWORD: 123456
    ports: ["5432:5432"]

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

# 启动
docker-compose -f docker/docker-compose.yml up -d