最佳实践

下面给你一份真正可落地的 Python FastAPI 最佳实践大全，结合你现有的技术栈（PostgreSQL、SQLAlchemy ORM、JWT、Casbin、UMI、AntD、Python≥3.12），全部内容可直接用于企业级项目。

整理成结构化的 30 条黄金准则，涵盖：项目结构、配置管理、数据库、权限、依赖注入、路由、响应格式、错误处理、工具函数、性能优化、部署经验等。

1. 项目结构最佳实践（目录）

✅ 分层清晰：API、Service、Repository、Schema、Core、DB

app/
 ├── api/                # 路由层：输入输出、权限校验
 │    ├── auth.py
 │    ├── users.py
 │    └── roles.py
 ├── services/           # 业务层：逻辑处理
 │    └── users.py
 ├── repositories/       # 数据层：DB CURD
 │    └── users.py
 ├── schemas/            # Pydantic 模型
 │    └── user.py
 ├── db/
 │    ├── session.py     # engine, async_sessionmaker
 │    ├── base.py
 │    └── models/
 ├── core/
 │    ├── config.py      # 统一配置
 │    ├── security.py    # JWT + bcrypt
 │    ├── casbin.py      # 权限
 │    └── middleware.py  # 日志、CORS、异常
 ├── utils/
 │    ├── json.py        # ORM 2 JSON
 │    └── pagination.py
 ├── main.py
 └── __init__.py

2. 使用 pydantic-settings 管理配置

pip install pydantic-settings

# core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    env: str = "dev"
    db_url: str
    jwt_secret: str
    class Config:
        env_file = ".env"

settings = Settings()

保持开发/生产隔离。

3. 必须使用 Async SQLAlchemy 2.0（你已经在用）

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine(settings.db_url)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

4. 统一依赖注入

async def get_db():
    async with SessionLocal() as session:
        yield session

减少全局变量污染。

5. Repository 模式（数据层）

每个模型一个 Repository
所有 CURD 集中维护

class UserRepo:
    @staticmethod
    async def get_by_id(db, id):
        return await db.get(User, id)

业务层(service)不应该直接操作 ORM。

6. Service 层：只做业务逻辑

class UserService:
    async def create_user(data, db):
        hashed = hash_password(data.password)
        user = User(**data.model_dump(), password=hashed)
        await UserRepo.save(db, user)
        return user

保证清晰职责划分。

7. 路由层：只做输入/输出与权限校验

@router.post("/")
async def create(user: UserCreate, db=Depends(get_db), u=Depends(check_perm("users", "create"))):
    return await UserService.create_user(user, db)

8. 统一响应格式（非常重要）

def ok(data=None, msg="success"):
    return {"code": 0, "msg": msg, "data": data}

def err(msg="error", code=400):
    return {"code": code, "msg": msg, "data": None}

前端（UMI）更容易处理。

9. 中间件统一处理异常

@app.exception_handler(Exception)
async def common_exception(request, exc):
    return JSONResponse(err("服务器异常", 500))

避免每个接口写 try/except。

10. JWT + bcrypt（你已经在用）

密码必须 bcrypt 存储：

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

JWT 推荐：

access_token 1 小时
refresh_token 7 天
需要 refresh 接口

11. JWT + Casbin 做极细权限控制（你的需求）

用户登录后加载权限
前端缓存权限
后端路由用 Decorator 校验

例如权限：“module:action:path:platform”

def check_perm(obj, act):
    def wrapper(user=Depends(get_current_user)):
        if not enforcer.enforce(user.role, obj, act):
            raise HTTPException(403, "权限不足")
        return user
    return wrapper

12. 数据库模型最佳实践（SQLAlchemy 2.0）

使用 mapped_column
使用 UUID 或 Identity
字段类型完整定义
创建联合索引与唯一约束（你已经在用）

示例：

id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(100), unique=True)

13. 必须定义 BaseModel 的 from_attributes

class UserOut(BaseModel):
    class Config:
        from_attributes = True

否则 ORM -> schema 会出错。

14. 统一 ORM -> JSON 工具（你一直在完善）

你的最佳版本：

def to_dict_list(items, fields=None):
    ...

并自动处理：

datetime -> str
strip()
忽略 SQLAlchemy 内部字段

这是最佳实践。

15. 分页最佳实践（全局复用）

def paginate(query, page: int, size: int):
    return query.offset((page - 1) * size).limit(size)

16. 路由 prefix 必须统一

app.include_router(user_router, prefix="/api/users", tags=["users"])

tags 让 Swagger 更清晰。

17. 使用 ORJSONResponse（加速 30%）

from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

18. 配置 CORS

app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"])

19. 日志使用 loguru

按日期分割
JSON 格式
接口自动记录请求/响应

20. 上传文件必须限制大小

添加中间件：

@app.middleware("http")
async def limit_upload(request, call_next):
    if request.url.path.startswith("/upload") and request.headers.get("content-length"):
        if int(request.headers["content-length"]) > 10 * 1024 * 1024:
            return err("文件太大", 413)
    return await call_next(request)

21. 应避免长阻塞，使用后台任务

from fastapi import BackgroundTasks

background.add_task(send_email, user.email)

22. 使用 lifespan 加载 Casbin、配置等

@app.on_event("startup")
async def start():
    init_casbin()

23. 路径设计使用 RESTful

功能	Method	Path
获取列表	GET	/users
创建	POST	/users
获取详情	GET	/users/{id}
删除	DELETE	/users/{id}

24. 严格的响应模型

@router.get("/", response_model=list[UserOut])

25. 使用 HTTP 状态码语义清晰

200 正常
201 创建成功
400 参数错误
401 未登录
403 权限不足
404 不存在
500 服务端错误

26. 限制数据库 Session 生命周期

一定不能复用 Global Session。

27. 数据库连接池配置

engine = create_async_engine(
    url,
    pool_size=20,
    max_overflow=40,
    pool_timeout=30,
)

28. 生产环境使用 Gunicorn + Uvicorn

gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4

29. 使用 Nginx 做反向代理 + HTTPS

推荐开启：

http2
gzip
cache-control

30. 自动化文档分为两套

/docs -> Swagger（面向前端）
/redoc -> ReDoc（面向产品、测试）