fastapi

Name: fastapi
Author: youlaitech

github.com/youlaitechDevelopment6348 installsUpdated 1w ago

About fastapi

A FastAPI backend development guide for developers building async REST APIs and admin modules. It specifies a stack, project structure, naming, API conventions, error handling, responses, ORM patterns, and authentication guidance.

Key features

01Async FastAPI and SQLAlchemy conventions
02Domain-oriented project structure
03Naming rules for modules and schemas
04REST routes and response standards
05HTTP and business error-code handling
06ORM, authentication, and testing guidance

Use cases

01Design a new FastAPI backend layout
02Implement consistent CRUD endpoints
03Standardize API errors and responses
04Review async ORM and validation patterns

Install

Copy the command and run it in a terminal.

$npx skills add https://github.com/youlaitech/youlai-skills --skill fastapi

After installing, type/fastapiin Claude Code to confirm it loaded.

Details

From the project README

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: 技术栈

层	选型	说明
Web 框架	FastAPI 0.115+	异步原生、Pydantic v2 集成、自动 OpenAPI 文档
ASGI 服务器	Uvicorn 0.30+	FastAPI 官方推荐
ORM	SQLAlchemy 2.0 + asyncpg	`Mapped` + `mapped_column` 声明式、完整 async 支持
数据验证	Pydantic v2	Rust 核心、FastAPI 原生集成
数据库	PostgreSQL 16+	生产级、asyncpg 异步驱动
缓存	Redis 7.x	`redis[hiredis]` 异步客户端
认证	PyJWT 2.x + bcrypt 4.x	JWT 签发/验签、密码哈希
日志	loguru 0.7+	零配置、彩色输出、自动 rotate
包管理	uv	Rust 编写、10-100x pip 速度
代码质量	ruff + mypy	Lint + 格式化 + 静态类型检查
测试	pytest + httpx	httpx AsyncClient 直测 FastAPI

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 文件命名

类型	规范	示例
模块目录	小写单数名词	`user/`, `role/`, `menu/`
Schema 文件	`schemas.py`	`app/system/user/schemas.py`
Service 文件	`service.py`	`app/auth/service.py`
Router 文件	`router.py`	`app/system/role/router.py`
配置文件	`config.py`	`app/config.py`
模型文件	按域拆分	`app/system/user/models.py`
常量文件	`constants.py`	`app/constants.py`（全局）/ `app/system/user/constants.py`（域）
枚举文件	`enums.py` / `constants.py`	`app/enums.py`（全局）/ `app/system/role/constants.py`（域枚举）
测试文件	`test_<模块>.py`	`tests/auth/test_auth.py`

3.2 类命名

类型	规范	示例
ORM 模型	`Sys` + PascalCase	`SysUser`, `SysRole`, `SysMenu`, `SysDept`
Pydantic Schema	功能 + 后缀	`UserCreate`, `UserUpdate`, `UserVO`, `UserQuery`
Service 类	模块名 + `Service`	`UserService`, `AuthService`, `RoleService`
Config 类	`Settings`	`Settings`（单例）
Enum 类	功能名 + `Enum`	`StatusEnum`, `DataScopeEnum`
Exception	`BusinessException`	统一使用 `BusinessException`

Schema 后缀规范：

后缀	用途	示例
`Create`	创建表单	`UserCreate`, `RoleCreate`
`Update`	更新表单	`UserUpdate`, `MenuUpdate`
`VO`	视图对象（返回）	`UserVO`, `RoleVO`, `DictVO`
`Query`	分页查询参数	`UserQuery`, `RoleQuery`, `LogQuery`
`Form`	特定操作表单	`UserPasswordForm`, `RoleMenuForm`
`Result`	响应结果	`LoginResult`, `ExcelResultVO`

3.3 方法命名

操作	方法名	示例
分页查询	`get_page`	`async def get_page(self, query: UserQuery) -> PageResult`
单条查询	`get_by_id`	`async def get_by_id(self, id: int) -> UserVO`
树形查询	`get_tree`	`async def get_tree(self, keywords: str) -> list[MenuVO]`
下拉选项	`get_options`	`async def get_options(self) -> list[dict]`
创建	`create`	`async def create(self, form: UserCreate) -> UserVO`
更新	`update`	`async def update(self, form: UserUpdate) -> UserVO`
删除（批量）	`delete`	`async def delete(self, ids: str) -> int`
状态变更	`update_status`	`async def update_status(self, id: int, status: int) -> None`
导入	`import_<资源>`	`async def import_users(self, data: list) -> dict`
导出	`export_<资源>`	`async def export_users(self, query) -> list[dict]`

3.4 变量命名

类型	规范	示例
变量	snake_case	`user_list`, `role_codes`
常量	UPPER_SNAKE_CASE	`MAX_PAGE_SIZE`, `ROOT_ROLE_CODE`
私有方法	`_` 前缀	`def _to_vo(self, m: SysUser) -> UserVO`
布尔变量	`is_`/`has_` 前缀	`is_root`, `is_deleted`, `has_perm`
数据库字段	snake_case	`dept_id`, `create_time`
JSON/API 字段	camelCase	`deptId`, `createTime`, `pageSize`

关键约定：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=["用户管理"])

模块	前缀
认证	`/api/v1/auth`
用户	`/api/v1/users`
角色	`/api/v1/roles`
菜单	`/api/v1/menus`
部门	`/api/v1/depts`
字典	`/api/v1/dict`
配置	`/api/v1/configs`
通知	`/api/v1/notices`
日志	`/api/v1/logs`
文件	`/api/v1/files`
SSE	`/api/v1/sse`

4.2 标准 CRUD 路径

操作	方法	路径	说明
分页列表	`GET`	`/api/v1/users`	Query 参数：`pageNum`/`pageSize`/`keywords`/...
详情	`GET`	`/api/v1/users/{user_id}`	路径参数 `user_id: int`
新增	`POST`	`/api/v1/users`	Body: `UserCreate`
更新	`PUT`	`/api/v1/users/{user_id}`	Body: `UserUpdate`
批量删除	`DELETE`	`/api/v1/users/{ids}`	`ids` 逗号分隔
修改状态	`PATCH`	`/api/v1/users/status`	Body: `UserStatusForm`
重置密码	`PATCH`	`/api/v1/users/password`	Body: `UserPasswordForm`
导出	`GET`	`/api/v1/users/export`	返回 StreamingResponse xlsx
导入	`POST`	`/api/v1/users/import`	UploadFile xlsx
导入模板	`GET`	`/api/v1/users/template`	下载模板 xlsx
下拉选项	`GET`	`/api/v1/roles/options`	返回 `[{value, label}]`

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 前端拦截器的双轨制方案。

方案	代表	缺陷
统一 200	部分国内项目、GraphQL	违背 HTTP 语义；网关/监控无法识别错误；API 工具误判成功
纯 HTTP 状态码	RFC 9457、Zuplo 最佳实践	无法表达细粒度业务错误（如"用户名不存在" vs "密码错误"都是 401）
双轨制（本项目）	youlai-fastapi	—

双轨制原则：

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

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

5.2 HTTP 状态码使用规范

HTTP 状态	含义	使用场景	body 中的 code 示例
`200`	成功	业务操作成功	`00000`
`400`	客户端请求错误	请求格式错误、参数缺失	`A0400`
`401`	未认证	Token 无效/过期、未登录	`A0230`、`A0231`
`403`	无权限	已认证但无操作权限	`A0301`
`404`	资源不存在	查询的数据不存在	`B0003`
`409`	冲突	唯一键冲突、状态冲突	`B0002`
`422`	参数校验失败	Pydantic 校验失败	`A0400`
`500`	服务器内部错误	未捕获异常、系统错误	`B0001`

为什么不用 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 错误码设计规范

首位	含义	HTTP 大类	范围
`0`	成功	2xx	`00000`
`A`	用户端错误	4xx	`A0001-A0999`
`B`	系统端错误	5xx	`B0001-B0999`
`C`	第三方服务错误	5xx	`C0001-C0999`

第二位子分类（A 类）：

第二位	子类	示例
`0`	通用	`A0001` 未知错误
`2`	认证/Token	`A0230` Token 过期、`A0231` Refresh 失效
`3`	授权/权限	`A0301` 权限不足
`4`	参数校验	`A0400` 校验失败、`A0401` 用户不存在、`A0402` 密码错误
`5`	租户	`A0250` 需要选择租户

新增错误码流程：

在 ResultCode 枚举中添加新值
在 http_status property 的 mapping 中添加映射
更新本规范的 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 字段命名对照

SQL 列名	ORM 字段	Schema 字段	说明
`dept_id`	`dept_id`	`deptId`	部门ID
`create_time`	`create_time`	`createTime`	创建时间
`is_deleted`	`is_deleted`	—	逻辑删除（内部使用）
`parent_id`	`parent_id`	`parentId`	父节点ID

转换规则：ORM → VO 时，to_vo 或 _to_vo 方法负责 snake_case → 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 环境变量切换：

值	模式	TokenManager
`jwt`（默认）	JWT 自包含 Token	`JwtTokenManager`
`redis-token`	Redis 存储 Token	`RedisTokenManager`

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

操作	说明
`list`	列表/查询
`detail`	详情
`create`	新增
`edit`	编辑
`delete`	删除
`import`	导入
`export`	导出
`reset-pwd`	重置密码
`assign-menu`	分配菜单权限

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 各层注释要点

层	必须说明	示例
模块（文件头）	文件职责	`"""字典项路由。"""`
类	职责、关键依赖	`"""字典项管理服务，依赖 AsyncSession。"""`
公共方法	功能、参数、返回值、异常	见 11.2
私有方法（`_` 前缀）	简要说明即可	`"""ORM 模型 → 视图对象。"""`

11.4 常见反模式

反模式	问题	正确做法
`def get_user(id): pass` 无 docstring	公共方法无文档	补充 docstring
`"""获取用户"""` 参数无说明	缺少 Args/Returns	补充完整 Args/Returns/Raises
docstring 与代码不一致	误导维护者	代码改动时同步更新
注释掉的代码块保留	历史代码可从 Git 查阅	直接删除
`# TODO fix this`	无标记人、无时间	`# TODO(zhangsan, 2025/06/01, 下版本修复)`

11.5 TODO / FIXME 标记规范

标记	用途	格式	示例
`TODO`	待实现	`TODO(标记人, 日期, [说明])`	`TODO(zhangsan, 2025/06/01, 扩展邮箱验证码)`
`FIXME`	已知 bug	`FIXME(标记人, 日期, [说明])`	`FIXME(lisi, 2025/06/15, 并发场景偶发死锁)`

Part 12: 代码质量检查清单

11.1 通用检查

11.2 数据库相关

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

11.3 API 相关

11.4 Service 层

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

Part 13: 常见反模式

反模式	问题	正确做法
`print("debug")`	无日志级别、无时间戳	`logger.info()` / `logger.debug()`
`result.scalars().first()`	不区分 None vs 第一条	`result.scalar_one_or_none()`
`model.dict()`	Pydantic v2 已废弃	`model.model_dump()`
硬编码配置	不便运维	用 `Settings` + `.env`
`db.execute()` 后不 `await`	异步未执行	`await self.db.execute(...)`
`flush` 前取 `obj.id`	自增 id 尚未生成	先 `flush()` 再取 `id`
SQL 查询缺 `is_deleted = 0`	查到已删除数据	所有查询加逻辑删除条件
固定路径在 `/{id}` 之后	先被路径参数拦截	固定路径注册在前
重复 import 同一模块	不必要	合并为单条 import
`try/except` 捕获所有异常后 pass	隐藏错误	只捕获已知异常，记录日志
所有错误统一返回 HTTP 200	网关/监控无法识别错误；违背 HTTP 语义	HTTP 状态码语义化 + body 中 `code` 双轨制（Part 5）
业务错误硬编码 HTTP 状态码	如 `JSONResponse(status_code=401)` 散落各处	用 `ResultCode.http_status` property 统一映射
字符串硬编码业务码	如 `code="A0230"`	用 `ResultCode.TOKEN_INVALID.value` 枚举

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

Comments

Markdown supported

Details

Author: youlaitech
Repository: youlaitech/youlai-skills
Category: Development
Installs: 48
GitHub stars: 63

Report

More skills

All skills

remotion-render

github.com/inference-sh

1.6M

Render videos from React/Remotion component code via inference.sh. Pass TSX code, get MP4. Supports all Remotion APIs: useCurrentFrame, useVideoConfig, spring, interpolate, AbsoluteFill, Sequence. Configurable resolution, FPS, duration, codec. Use for: programmatic video generation, animated graphics, motion design, data-driven videos, React animations to video. Triggers: remotion, render video from code, tsx to video, react video, programmatic video, remotion render, code to video, animated video, motion graphics code, react animation video

improve-codebase-architecture

github.com/mattpocock

1.2M

Scan a codebase for architecture improvement opportunities, present them as a visual HTML report, then dive deep into whichever one you pick.

design-mobile-apps

github.com/designed-by-ai

1.2M

Design mobile app screens with Sleek, edit Sleek projects, and implement their designs in React Native or HTML

vercel-react-best-practices

github.com/vercel-labs

860.4k

React and Next.js performance optimization guidelines from Vercel Engineering. This skill should be used when writing, reviewing, or refactoring React/Next.js code to ensure optimal performance patterns. Triggers on tasks involving React components, Next.js pages, data fetching, bundle optimization, or performance improvements.

fastapi

About fastapi

Key features

Use cases

Install

Details

FastAPI 后端开发规范

触发条件

Part 1: 技术栈

Part 2: 目录结构

Part 3: 命名规范

3.1 文件命名

3.2 类命名

3.3 方法命名

3.4 变量命名

3.5 import 排序

Part 4: 路由与 API 规范

4.1 路由前缀

4.2 标准 CRUD 路径

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

4.4 Router 函数模板

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

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

5.2 HTTP 状态码使用规范

5.3 业务错误码（ResultCode）

5.4 BusinessException 与全局异常处理器

5.5 业务异常抛出示例

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

5.7 错误码设计规范

Part 6: 统一响应格式

6.1 Result 结构

6.2 PageResult 结构

6.3 分页查询参数

6.4 返回示例

Part 7: ORM 模型规范

7.1 基类 Mixin

7.2 ORM 模型示例

7.3 字段命名对照

7.4 VO 转换模板

Part 8: 认证与鉴权

8.1 Token 模式

8.2 获取当前用户

8.3 权限校验

8.4 权限标识命名规范

Part 9: 数据库操作规范

9.1 异步查询

9.2 事务管理

9.3 批量操作

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

Step 1: 检查 ORM 模型

Step 2: 创建模块目录

Step 3: 创建 schemas.py

Step 4: 创建 service.py

Step 5: 创建 router.py

Step 6: 创建 __init__.py

Step 7: 在 main.py 注册路由

Step 8: 验证

模块文件清单

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

Part 11: 注释规范

11.1 总则

11.2 docstring 格式（Google 风格）

11.3 各层注释要点

11.4 常见反模式

11.5 TODO / FIXME 标记规范

Part 12: 代码质量检查清单

11.1 通用检查

11.2 数据库相关

11.3 API 相关

11.4 Service 层

Part 13: 常见反模式

Part 14: 单模块文件

Part 15: Docker 部署

Comments

Step 3: 创建 `schemas.py`

Step 4: 创建 `service.py`

Step 5: 创建 `router.py`

Step 6: 创建 `init.py`

Step 7: 在 `main.py` 注册路由