fastapi-expert

FastAPI Expert

精通异步 Python、Pydantic V2 以及基于 FastAPI 的生产级 API 开发。

何时使用此 Skill

• 基于 FastAPI 构建 REST API
• 实现 Pydantic V2 数据校验 Schema
• 配置异步数据库操作
• 实现 JWT 身份认证与权限控制
• 创建 WebSocket 端点
• 优化 API 性能

核心工作流

1. 分析需求 — 明确 API 端点、数据模型及认证授权需求
2. 设计 Schema — 创建用于数据校验的 Pydantic V2 模型
3. 编码实现 — 编写异步端点，并合理运用依赖注入机制
4. 加固安全 — 添加身份认证、权限控制及限流策略
5. 单元测试 — 使用 pytest 和 httpx 编写异步测试；每完成一组端点开发后运行 pytest，并在 /docs 验证 OpenAPI 文档

每步检查点： 在继续下一步之前，确认 Schema 校验正常、端点返回预期 HTTP 状态码，且 /docs 准确反映了设计的 API 结构。

最小完整示例

整合 Schema、端点与依赖注入的完整最小单元示例：

# schemas.py
from pydantic import BaseModel, EmailStr, field_validator, model_config

class UserCreate(BaseModel):
    model_config = model_config(str_strip_whitespace=True)

    email: EmailStr
    password: str
    name: str | None = None

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

class UserResponse(BaseModel):
    model_config = model_config(from_attributes=True)

    id: int
    email: EmailStr
    name: str | None = None

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated

from app.database import get_db
from app.schemas import UserCreate, UserResponse
from app import crud

router = APIRouter(prefix="/users", tags=["users"])

DbDep = Annotated[AsyncSession, Depends(get_db)]

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: DbDep) -> UserResponse:
    existing = await crud.get_user_by_email(db, payload.email)
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
    return await crud.create_user(db, payload)

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from app.schemas import UserCreate
from app.security import hash_password

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

async def create_user(db: AsyncSession, payload: UserCreate) -> User:
    user = User(email=payload.email, hashed_password=hash_password(payload.password), name=payload.name)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

JWT 身份认证代码片段

# security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

SECRET_KEY = "read-from-env"  # use os.environ / settings
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_access_token(subject: str, expires_delta: timedelta = timedelta(minutes=30)) -> str:
    payload = {"sub": subject, "exp": datetime.now(timezone.utc) + expires_delta}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
    try:
        data = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        subject: str | None = data.get("sub")
        if subject is None:
            raise ValueError
        return subject
    except (JWTError, ValueError):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

CurrentUser = Annotated[str, Depends(get_current_user)]

参考指南

根据具体上下文加载详细指南：

主题	参考文档	加载时机
Pydantic V2	references/pydantic-v2.md	创建 Schema、数据校验、配置 model_config 时
SQLAlchemy	references/async-sqlalchemy.md	进行异步数据库操作、定义模型、编写 CRUD 时
Endpoints	references/endpoints-routing.md	使用 APIRouter、配置依赖注入、路由规划时
Authentication	references/authentication.md	实现 JWT、OAuth2、get_current_user 时
Testing	references/testing-async.md	使用 pytest-asyncio、httpx、编写 Fixtures 时
Django Migration	references/migration-from-django.md	从 Django/DRF 迁移至 FastAPI 时

开发规范

强制要求

• 全局使用类型提示（FastAPI 强依赖类型声明）
• 使用 Pydantic V2 语法（field_validator、model_validator、model_config）
• 依赖注入统一使用 Annotated 模式
• 所有 I/O 操作均使用 async/await
• 使用 X | None 替代 Optional[X]
• 返回规范的 HTTP 状态码
• 规范补充端点文档（用于自动生成 OpenAPI）

严禁事项

• 严禁使用同步数据库操作
• 严禁绕过 Pydantic 数据校验
• 严禁明文存储密码
• 严禁在响应数据中泄漏敏感信息
• 严禁使用 Pydantic V1 语法（如 @validator、class Config）
• 严禁不规范地混用同步与异步代码
• 严禁硬编码配置项

输出模板

实现 FastAPI 功能时，需提供以下内容：

1. Schema 文件（Pydantic 数据模型）
2. 端点文件（包含路由与 endpoint 的 router）
3. CRUD 操作代码（若涉及数据库交互）
4. 核心设计决策的简要说明

知识参考

FastAPI, Pydantic V2, async SQLAlchemy, Alembic 数据库迁移, JWT/OAuth2, pytest-asyncio, httpx, BackgroundTasks, WebSockets, 依赖注入, OpenAPI/Swagger

文档