FastAPI 生产级模板：Pydantic v2 + SQLAlchemy 异步 + JWT 身份验证

fastapi by srbhr/resume-matcher

175 周安装量

26,500 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/srbhr/resume-matcher --skill fastapi

Python Web框架开发 API

🇨🇳中文介绍

FastAPI Skill

使用 Pydantic v2、SQLAlchemy 2.0 异步版本和 JWT 身份验证的经过生产环境测试的 FastAPI 模式。

最新版本（2026年1月验证）：

FastAPI: 0.128.0
Pydantic: 2.11.7
SQLAlchemy: 2.0.30
Uvicorn: 0.35.0
python-jose: 3.3.0

要求：

Python 3.9+（FastAPI 0.125.0 中放弃了对 Python 3.8 的支持）
Pydantic v2.7.0+（FastAPI 0.128.0 中完全移除了对 Pydantic v1 的支持）

快速开始

使用 uv 设置项目

# 创建项目
uv init my-api
cd my-api

# 添加依赖
uv add fastapi[standard] sqlalchemy[asyncio] aiosqlite python-jose[cryptography] passlib[bcrypt]

# 运行开发服务器
uv run fastapi dev src/main.py

最小工作示例

# src/main.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="My API")

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.post("/items")
async def create_item(item: Item):
    return item

运行：uv run fastapi dev src/main.py

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

项目结构（基于领域）

为了项目可维护性，请按领域而非文件类型组织：

my-api/
├── pyproject.toml
├── src/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用初始化
│   ├── config.py            # 全局设置
│   ├── database.py          # 数据库连接
│   │
│   ├── auth/                # 身份验证领域
│   │   ├── __init__.py
│   │   ├── router.py        # 身份验证端点
│   │   ├── schemas.py       # Pydantic 模型
│   │   ├── models.py        # SQLAlchemy 模型
│   │   ├── service.py       # 业务逻辑
│   │   └── dependencies.py  # 身份验证依赖项
│   │
│   ├── items/               # 项目领域
│   │   ├── __init__.py
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   └── service.py
│   │
│   └── shared/              # 共享工具
│       ├── __init__.py
│       └── exceptions.py
└── tests/
    └── test_main.py

Pydantic 模式（验证）

# src/items/schemas.py
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from enum import Enum

class ItemStatus(str, Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be positive")
    status: ItemStatus = ItemStatus.DRAFT

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None
    price: float | None = Field(None, gt=0)
    status: ItemStatus | None = None

class ItemResponse(ItemBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

使用 Field() 进行验证约束
分离创建/更新/响应模式
from_attributes=True 启用 SQLAlchemy 模型转换
使用 str | None（Python 3.10+）而非 Optional[str]

SQLAlchemy 模型（数据库）

# src/items/models.py
from sqlalchemy import String, Float, DateTime, Enum as SQLEnum
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from src.database import Base
from src.items.schemas import ItemStatus

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
    price: Mapped[float] = mapped_column(Float)
    status: Mapped[ItemStatus] = mapped_column(
        SQLEnum(ItemStatus), default=ItemStatus.DRAFT
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

数据库设置（异步 SQLAlchemy 2.0）

# src/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "sqlite+aiosqlite:///./database.db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# src/items/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.items import schemas, models

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

@router.get("", response_model=list[schemas.ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.Item).offset(skip).limit(limit)
    )
    return result.scalars().all()

@router.get("/{item_id}", response_model=schemas.ItemResponse)
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(models.Item).where(models.Item.id == item_id)
    )
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("", response_model=schemas.ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
    item_in: schemas.ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.database import engine, Base
from src.items.router import router as items_router
from src.auth.router import router as auth_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动：创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭：如有需要，进行清理

app = FastAPI(title="My API", lifespan=lifespan)

# CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 您的前端
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 包含路由器
app.include_router(auth_router)
app.include_router(items_router)

# src/auth/schemas.py
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    email: str

    model_config = ConfigDict(from_attributes=True)

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int | None = None

# src/auth/service.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from src.config import settings

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

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

身份验证依赖项

# src/auth/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import service, models, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> models.User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    payload = service.decode_token(token)
    if payload is None:
        raise credentials_exception

    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception

    result = await db.execute(
        select(models.User).where(models.User.id == int(user_id))
    )
    user = result.scalar_one_or_none()

    if user is None:
        raise credentials_exception

    return user

身份验证路由器

# src/auth/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import schemas, models, service
from src.auth.dependencies import get_current_user

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

@router.post("/register", response_model=schemas.UserResponse)
async def register(
    user_in: schemas.UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # 检查是否已存在
    result = await db.execute(
        select(models.User).where(models.User.email == user_in.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(
        email=user_in.email,
        hashed_password=service.hash_password(user_in.password)
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.post("/login", response_model=schemas.Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.User).where(models.User.email == form_data.username)
    )
    user = result.scalar_one_or_none()

    if not user or not service.verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    access_token = service.create_access_token(data={"sub": str(user.id)})
    return schemas.Token(access_token=access_token)

@router.get("/me", response_model=schemas.UserResponse)
async def get_me(current_user: models.User = Depends(get_current_user)):
    return current_user

# 在任何路由器中
from src.auth.dependencies import get_current_user
from src.auth.models import User

@router.post("/items")
async def create_item(
    item_in: schemas.ItemCreate,
    current_user: User = Depends(get_current_user),  # 需要身份验证
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump(), user_id=current_user.id)
    # ...

# src/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = "sqlite+aiosqlite:///./database.db"
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

DATABASE_URL=sqlite+aiosqlite:///./database.db
SECRET_KEY=your-super-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=30

将 Pydantic 模式与 SQLAlchemy 模型分离 - 不同职责，不同文件
对 I/O 操作使用异步 - 数据库、HTTP 调用、文件访问
使用 Pydantic Field() 进行验证 - 约束、默认值、描述
使用依赖注入 - Depends() 用于数据库、身份验证、验证
返回适当的状态码 - 创建返回 201，删除返回 204 等。

切勿在异步路由中使用阻塞调用 - 不要使用 time.sleep()，使用 asyncio.sleep()
切勿将业务逻辑放在路由中 - 使用服务层
切勿硬编码密钥 - 使用环境变量
切勿跳过验证 - 始终使用 Pydantic 模式
切勿在生产环境中在 CORS 来源中使用 * - 指定确切的来源

此技能可预防官方 FastAPI GitHub 和发布说明中记录的 7 个问题。

问题 #1：表单数据丢失字段设置元数据

错误：使用 Form() 时，model.model_fields_set 包含默认值来源：GitHub Issue #13399 原因：表单数据解析会预加载默认值并将其传递给验证器，使得无法区分用户显式设置的字段和使用默认值的字段。此错误仅影响表单数据，不影响 JSON 请求体数据。

# ✗ 避免：当您需要 field_set 元数据时，使用带有 Form 的 Pydantic 模型
from typing import Annotated
from fastapi import Form

@app.post("/form")
async def endpoint(model: Annotated[MyModel, Form()]):
    fields = model.model_fields_set  # 不可靠！ ❌

# ✓ 使用：单独的 form 字段或 JSON 请求体
@app.post("/form-individual")
async def endpoint(
    field_1: Annotated[bool, Form()] = True,
    field_2: Annotated[str | None, Form()] = None
):
    # 您确切知道提供了什么 ✓

# ✓ 或者：当元数据重要时，使用 JSON 请求体
@app.post("/json")
async def endpoint(model: MyModel):
    fields = model.model_fields_set  # 正常工作 ✓

问题 #2：自定义响应静默覆盖 BackgroundTasks

错误：通过 BackgroundTasks 依赖项添加的后台任务不运行来源：GitHub Issue #11215 原因：当您返回带有 background 参数的自定义 Response 时，它会覆盖所有添加到注入的 BackgroundTasks 依赖项中的任务。这没有文档说明，并导致静默失败。

# ✗ 错误：混合使用两种机制
from fastapi import BackgroundTasks
from starlette.responses import Response, BackgroundTask

@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)  # 这个会丢失！ ❌
    return Response(
        content="Done",
        background=BackgroundTask(log_event)  # 只有这个会运行
    )

# ✓ 正确：仅使用 BackgroundTasks 依赖项
@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)
    tasks.add_task(log_event)
    return {"status": "done"}  # 所有任务都会运行 ✓

# ✓ 或者：仅使用 Response 的 background 参数（但无法注入依赖项）
@app.get("/")
async def endpoint():
    return Response(
        content="Done",
        background=BackgroundTask(log_event)
    )

规则：选择一种机制并坚持使用。不要混合使用注入的 BackgroundTasks 和 Response(background=...)。

问题 #3：可选表单字段在 TestClient 中中断（回归）

错误：对于可选的 Literal 字段，出现 422: "Input should be 'abc' or 'def'" 来源：GitHub Issue #12245 原因：从 FastAPI 0.114.0 开始，带有 Literal 类型的可选表单字段在通过 TestClient 传递 None 时验证失败。在 0.113.0 中正常工作。

from typing import Annotated, Literal, Optional
from fastapi import Form
from fastapi.testclient import TestClient

# ✗ 有问题：带有 Form 的可选 Literal（在 0.114.0+ 中中断）
@app.post("/")
async def endpoint(
    attribute: Annotated[Optional[Literal["abc", "def"]], Form()]
):
    return {"attribute": attribute}

client = TestClient(app)
data = {"attribute": None}  # 或者省略该字段
response = client.post("/", data=data)  # 返回 422 ❌

# ✓ 解决方法 1：不要显式传递 None，省略该字段
data = {}  # 省略而非 None
response = client.post("/", data=data)  # 工作 ✓

# ✓ 解决方法 2：避免在可选表单字段中使用 Literal 类型
@app.post("/")
async def endpoint(attribute: Annotated[str | None, Form()] = None):
    # 在应用逻辑中进行验证
    if attribute and attribute not in ["abc", "def"]:
        raise HTTPException(400, "Invalid attribute")

问题 #4：Pydantic Json 类型不适用于表单数据

错误："JSON object must be str, bytes or bytearray" 来源：GitHub Issue #10997 原因：直接将 Pydantic 的 Json 类型与 Form() 一起使用会失败。您必须将该字段作为 str 接收并手动解析。

from typing import Annotated
from fastapi import Form
from pydantic import Json, BaseModel

# ✗ 错误：Json 类型直接与 Form 一起使用
@app.post("/broken")
async def broken(json_list: Annotated[Json[list[str]], Form()]) -> list[str]:
    return json_list  # 返回 422 ❌

# ✓ 正确：作为 str 接收，用 Pydantic 解析
class JsonListModel(BaseModel):
    json_list: Json[list[str]]

@app.post("/working")
async def working(json_list: Annotated[str, Form()]) -> list[str]:
    model = JsonListModel(json_list=json_list)  # Pydantic 在此处解析
    return model.json_list  # 工作 ✓

问题 #5：带有前向引用的 Annotated 破坏 OpenAPI 生成

错误：依赖项类型的 OpenAPI 模式缺失或不正确来源：GitHub Issue #13056 原因：当使用带有 Depends() 和前向引用（来自 __future__ import annotations）的 Annotated 时，OpenAPI 模式生成失败或产生不正确的模式。

# ✗ 有问题：带有 Depends 的前向引用
from __future__ import annotations
from dataclasses import dataclass
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

def get_potato() -> Potato:  # 前向引用
    return Potato(color='red', size=10)

@app.get('/')
async def read_root(potato: Annotated[Potato, Depends(get_potato)]):
    return {'Hello': 'World'}
# OpenAPI 模式未正确包含 Potato 定义 ❌

@dataclass
class Potato:
    color: str
    size: int

# ✓ 解决方法 1：不要在路由文件中使用 __future__ 注解
# 移除：from __future__ import annotations

# ✓ 解决方法 2：对类型提示使用字符串字面量
def get_potato() -> "Potato":
    return Potato(color='red', size=10)

# ✓ 解决方法 3：在依赖项中使用之前定义类
@dataclass
class Potato:
    color: str
    size: int

def get_potato() -> Potato:  # 现在工作 ✓
    return Potato(color='red', size=10)

问题 #6：Pydantic v2 路径参数联合类型破坏性变更

错误：带有 int | str 的路径参数在 Pydantic v2 中始终解析为 str 来源：GitHub Issue #11251 | 社区来源原因：从 Pydantic v1 迁移到 v2 时的重大破坏性变更。路径/查询参数中带有 str 的联合类型现在始终解析为 str（在 v1 中正常工作）。

from uuid import UUID

# ✗ 有问题：路径参数中带有 str 的联合类型
@app.get("/int/{path}")
async def int_path(path: int | str):
    return str(type(path))
    # Pydantic v1: 对于 "123" 返回 <class 'int'>
    # Pydantic v2: 对于 "123" 返回 <class 'str'> ❌

@app.get("/uuid/{path}")
async def uuid_path(path: UUID | str):
    return str(type(path))
    # Pydantic v1: 对于有效的 UUID 返回 <class 'uuid.UUID'>
    # Pydantic v2: 返回 <class 'str'> ❌

# ✓ 正确：避免在路径/查询参数中使用带有 str 的联合类型
@app.get("/int/{path}")
async def int_path(path: int):
    return str(type(path))  # 正常工作 ✓

# ✓ 替代方案：如果需要类型转换，使用验证器
from pydantic import field_validator

class PathParams(BaseModel):
    path: int | str

    @field_validator('path')
    def coerce_to_int(cls, v):
        if isinstance(v, str) and v.isdigit():
            return int(v)
        return v

问题 #7：field_validator 中的 ValueError 返回 500 而非 422

错误：在自定义验证器中引发 ValueError 时出现 500 Internal Server Error 来源：GitHub Discussion #10779 | 社区来源原因：在带有表单字段的 Pydantic @field_validator 中引发 ValueError 时，FastAPI 返回 500 内部服务器错误，而不是预期的 422 不可处理实体验证错误。

from typing import Annotated
from fastapi import Form
from pydantic import BaseModel, field_validator, ValidationError, Field

# ✗ 错误：验证器中的 ValueError
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValueError("Value must be positive")  # 返回 500！ ❌
        return v

# ✓ 正确 1：改为引发 ValidationError
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValidationError("Value must be positive")  # 返回 422 ✓
        return v

# ✓ 正确 2：使用 Pydantic 的内置约束
class MyForm(BaseModel):
    value: Annotated[int, Field(gt=0)]  # 内置验证，返回 422 ✓

常见错误与修复

422 不可处理实体

原因：请求体与 Pydantic 模式不匹配

检查 /docs 端点 - 先在那里测试
验证 JSON 结构是否与模式匹配
检查必填字段与可选字段

修复：添加自定义验证错误处理器：

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

原因：缺少或配置错误的 CORS 中间件

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 生产环境中不要用 "*"
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

异步阻塞事件循环

原因：异步路由中的阻塞调用（例如 time.sleep()、同步数据库客户端、CPU 密集型操作）

症状（生产规模）：

吞吐量远早于预期达到平台期
随着并发增加，延迟“激增”
在高负载下，请求模式看起来几乎是串行的
当事件循环饱和时，请求无限排队
不明显的小型零散阻塞调用（非无限循环）

修复：使用异步替代方案：

# ✗ 错误：阻塞事件循环
import time
from sqlalchemy import create_engine  # 同步客户端

@app.get("/users")
async def get_users():
    time.sleep(0.1)  # 即使是小的阻塞在规模上也会累积！
    result = sync_db_client.query("SELECT * FROM users")  # 阻塞！
    return result

# ✓ 正确 1：使用异步数据库驱动
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    await asyncio.sleep(0.1)  # 非阻塞
    result = await db.execute(select(User))
    return result.scalars().all()

# ✓ 正确 2：对 CPU 密集型路由使用 def（而非 async def）
# FastAPI 自动在线程池中运行 def 路由
@app.get("/cpu-heavy")
def cpu_heavy_task():  # 注意：def 而非 async def
    return expensive_cpu_work()  # 在线程池中运行 ✓

# ✓ 正确 3：在异步路由中使用 run_in_executor 处理阻塞调用
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@app.get("/mixed")
async def mixed_task():
    # 在线程池中运行阻塞函数
    result = await asyncio.get_event_loop().run_in_executor(
        executor,
        blocking_function  # 您的阻塞函数
    )
    return result

来源：生产案例研究（2026年1月） | 社区来源

可选字段出现 "Field required"

原因：使用没有默认值的 Optional[str]

# 错误
description: Optional[str]  # 仍然是必填的！

# 正确
description: str | None = None  # 带有默认值的可选

# tests/test_main.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_root(client):
    response = await client.get("/")
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_create_item(client):
    response = await client.post(
        "/items",
        json={"name": "Test", "price": 9.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test"

运行：uv run pytest

uv run fastapi dev src/main.py

uv run uvicorn src.main:app --host 0.0.0.0 --port 8000

Gunicorn + Uvicorn（带工作进程的生产环境）

uv add gunicorn
uv run gunicorn src.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

FROM python:3.12-slim

WORKDIR /app
COPY . .

RUN pip install uv && uv sync

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

最后验证：2026-01-21 | 技能版本：1.1.0 | 变更：添加了 7 个已知问题（表单数据错误、后台任务、Pydantic v2 迁移陷阱），扩展了带有生产模式的异步阻塞指导 维护者：Jezweb | jeremy@jezweb.net

🇺🇸English

FastAPI Skill

Production-tested patterns for FastAPI with Pydantic v2, SQLAlchemy 2.0 async, and JWT authentication.

Latest Versions (verified January 2026):

FastAPI: 0.128.0
Pydantic: 2.11.7
SQLAlchemy: 2.0.30
Uvicorn: 0.35.0
python-jose: 3.3.0

Requirements :

Python 3.9+ (Python 3.8 support dropped in FastAPI 0.125.0)
Pydantic v2.7.0+ (Pydantic v1 support completely removed in FastAPI 0.128.0)

Quick Start

Project Setup with uv

# Create project
uv init my-api
cd my-api

# Add dependencies
uv add fastapi[standard] sqlalchemy[asyncio] aiosqlite python-jose[cryptography] passlib[bcrypt]

# Run development server
uv run fastapi dev src/main.py

Minimal Working Example

# src/main.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="My API")

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.post("/items")
async def create_item(item: Item):
    return item

Run: uv run fastapi dev src/main.py

Docs available at: http://127.0.0.1:8000/docs

Project Structure (Domain-Based)

For maintainable projects, organize by domain not file type:

my-api/
├── pyproject.toml
├── src/
│   ├── __init__.py
│   ├── main.py              # FastAPI app initialization
│   ├── config.py            # Global settings
│   ├── database.py          # Database connection
│   │
│   ├── auth/                # Auth domain
│   │   ├── __init__.py
│   │   ├── router.py        # Auth endpoints
│   │   ├── schemas.py       # Pydantic models
│   │   ├── models.py        # SQLAlchemy models
│   │   ├── service.py       # Business logic
│   │   └── dependencies.py  # Auth dependencies
│   │
│   ├── items/               # Items domain
│   │   ├── __init__.py
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   └── service.py
│   │
│   └── shared/              # Shared utilities
│       ├── __init__.py
│       └── exceptions.py
└── tests/
    └── test_main.py

Core Patterns

Pydantic Schemas (Validation)

# src/items/schemas.py
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from enum import Enum

class ItemStatus(str, Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be positive")
    status: ItemStatus = ItemStatus.DRAFT

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None
    price: float | None = Field(None, gt=0)
    status: ItemStatus | None = None

class ItemResponse(ItemBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

Key Points :

Use Field() for validation constraints
Separate Create/Update/Response schemas
from_attributes=True enables SQLAlchemy model conversion
Use str | None (Python 3.10+) not Optional[str]

SQLAlchemy Models (Database)

# src/items/models.py
from sqlalchemy import String, Float, DateTime, Enum as SQLEnum
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from src.database import Base
from src.items.schemas import ItemStatus

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
    price: Mapped[float] = mapped_column(Float)
    status: Mapped[ItemStatus] = mapped_column(
        SQLEnum(ItemStatus), default=ItemStatus.DRAFT
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

Database Setup (Async SQLAlchemy 2.0)

# src/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "sqlite+aiosqlite:///./database.db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Router Pattern

# src/items/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.items import schemas, models

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

@router.get("", response_model=list[schemas.ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.Item).offset(skip).limit(limit)
    )
    return result.scalars().all()

@router.get("/{item_id}", response_model=schemas.ItemResponse)
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(models.Item).where(models.Item.id == item_id)
    )
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("", response_model=schemas.ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
    item_in: schemas.ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

Main App

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.database import engine, Base
from src.items.router import router as items_router
from src.auth.router import router as auth_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Create tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: cleanup if needed

app = FastAPI(title="My API", lifespan=lifespan)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Your frontend
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(auth_router)
app.include_router(items_router)

JWT Authentication

Auth Schemas

# src/auth/schemas.py
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    email: str

    model_config = ConfigDict(from_attributes=True)

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int | None = None

Auth Service

# src/auth/service.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from src.config import settings

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

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

Auth Dependencies

# src/auth/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import service, models, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> models.User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    payload = service.decode_token(token)
    if payload is None:
        raise credentials_exception

    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception

    result = await db.execute(
        select(models.User).where(models.User.id == int(user_id))
    )
    user = result.scalar_one_or_none()

    if user is None:
        raise credentials_exception

    return user

Auth Router

# src/auth/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import schemas, models, service
from src.auth.dependencies import get_current_user

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

@router.post("/register", response_model=schemas.UserResponse)
async def register(
    user_in: schemas.UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # Check existing
    result = await db.execute(
        select(models.User).where(models.User.email == user_in.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(
        email=user_in.email,
        hashed_password=service.hash_password(user_in.password)
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.post("/login", response_model=schemas.Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.User).where(models.User.email == form_data.username)
    )
    user = result.scalar_one_or_none()

    if not user or not service.verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    access_token = service.create_access_token(data={"sub": str(user.id)})
    return schemas.Token(access_token=access_token)

@router.get("/me", response_model=schemas.UserResponse)
async def get_me(current_user: models.User = Depends(get_current_user)):
    return current_user

Protect Routes

# In any router
from src.auth.dependencies import get_current_user
from src.auth.models import User

@router.post("/items")
async def create_item(
    item_in: schemas.ItemCreate,
    current_user: User = Depends(get_current_user),  # Requires auth
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump(), user_id=current_user.id)
    # ...

Configuration

# src/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = "sqlite+aiosqlite:///./database.db"
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

Create .env:

DATABASE_URL=sqlite+aiosqlite:///./database.db
SECRET_KEY=your-super-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=30

Critical Rules

Always Do

Separate Pydantic schemas from SQLAlchemy models - Different jobs, different files
Use async for I/O operations - Database, HTTP calls, file access
Validate with Pydantic Field() - Constraints, defaults, descriptions
Use dependency injection - Depends() for database, auth, validation
Return proper status codes - 201 for create, 204 for delete, etc.

Never Do

Never use blocking calls in async routes - No time.sleep(), use asyncio.sleep()
Never put business logic in routes - Use service layer
Never hardcode secrets - Use environment variables
Never skip validation - Always use Pydantic schemas
Never use* in CORS origins for production - Specify exact origins

Known Issues Prevention

This skill prevents 7 documented issues from official FastAPI GitHub and release notes.

Issue #1: Form Data Loses Field Set Metadata

Error : model.model_fields_set includes default values when using Form() Source : GitHub Issue #13399 Why It Happens : Form data parsing preloads default values and passes them to the validator, making it impossible to distinguish between fields explicitly set by the user and fields using defaults. This bug ONLY affects Form data, not JSON body data.

Prevention :

# ✗ AVOID: Pydantic model with Form when you need field_set metadata
from typing import Annotated
from fastapi import Form

@app.post("/form")
async def endpoint(model: Annotated[MyModel, Form()]):
    fields = model.model_fields_set  # Unreliable! ❌

# ✓ USE: Individual form fields or JSON body instead
@app.post("/form-individual")
async def endpoint(
    field_1: Annotated[bool, Form()] = True,
    field_2: Annotated[str | None, Form()] = None
):
    # You know exactly what was provided ✓

# ✓ OR: Use JSON body when metadata matters
@app.post("/json")
async def endpoint(model: MyModel):
    fields = model.model_fields_set  # Works correctly ✓

Issue #2: BackgroundTasks Silently Overwritten by Custom Response

Error : Background tasks added via BackgroundTasks dependency don't run Source : GitHub Issue #11215 Why It Happens : When you return a custom Response with a background parameter, it overwrites all tasks added to the injected BackgroundTasks dependency. This is not documented and causes silent failures.

Prevention :

# ✗ WRONG: Mixing both mechanisms
from fastapi import BackgroundTasks
from starlette.responses import Response, BackgroundTask

@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)  # This will be lost! ❌
    return Response(
        content="Done",
        background=BackgroundTask(log_event)  # Only this runs
    )

# ✓ RIGHT: Use only BackgroundTasks dependency
@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)
    tasks.add_task(log_event)
    return {"status": "done"}  # All tasks run ✓

# ✓ OR: Use only Response background (but can't inject dependencies)
@app.get("/")
async def endpoint():
    return Response(
        content="Done",
        background=BackgroundTask(log_event)
    )

Rule : Pick ONE mechanism and stick with it. Don't mix injected BackgroundTasks with Response(background=...).

Issue #3: Optional Form Fields Break with TestClient (Regression)

Error : 422: "Input should be 'abc' or 'def'" for optional Literal fields Source : GitHub Issue #12245 Why It Happens : Starting in FastAPI 0.114.0, optional form fields with Literal types fail validation when passed None via TestClient. Worked in 0.113.0.

Prevention :

from typing import Annotated, Literal, Optional
from fastapi import Form
from fastapi.testclient import TestClient

# ✗ PROBLEMATIC: Optional Literal with Form (breaks in 0.114.0+)
@app.post("/")
async def endpoint(
    attribute: Annotated[Optional[Literal["abc", "def"]], Form()]
):
    return {"attribute": attribute}

client = TestClient(app)
data = {"attribute": None}  # or omit the field
response = client.post("/", data=data)  # Returns 422 ❌

# ✓ WORKAROUND 1: Don't pass None explicitly, omit the field
data = {}  # Omit instead of None
response = client.post("/", data=data)  # Works ✓

# ✓ WORKAROUND 2: Avoid Literal types with optional form fields
@app.post("/")
async def endpoint(attribute: Annotated[str | None, Form()] = None):
    # Validate in application logic instead
    if attribute and attribute not in ["abc", "def"]:
        raise HTTPException(400, "Invalid attribute")

Issue #4: Pydantic Json Type Doesn't Work with Form Data

Error : "JSON object must be str, bytes or bytearray" Source : GitHub Issue #10997 Why It Happens : Using Pydantic's Json type directly with Form() fails. You must accept the field as str and parse manually.

Prevention :

from typing import Annotated
from fastapi import Form
from pydantic import Json, BaseModel

# ✗ WRONG: Json type directly with Form
@app.post("/broken")
async def broken(json_list: Annotated[Json[list[str]], Form()]) -> list[str]:
    return json_list  # Returns 422 ❌

# ✓ RIGHT: Accept as str, parse with Pydantic
class JsonListModel(BaseModel):
    json_list: Json[list[str]]

@app.post("/working")
async def working(json_list: Annotated[str, Form()]) -> list[str]:
    model = JsonListModel(json_list=json_list)  # Pydantic parses here
    return model.json_list  # Works ✓

Issue #5: Annotated with ForwardRef Breaks OpenAPI Generation

Error : Missing or incorrect OpenAPI schema for dependency types Source : GitHub Issue #13056 Why It Happens : When using Annotated with Depends() and a forward reference (from __future__ import annotations), OpenAPI schema generation fails or produces incorrect schemas.

Prevention :

# ✗ PROBLEMATIC: Forward reference with Depends
from __future__ import annotations
from dataclasses import dataclass
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

def get_potato() -> Potato:  # Forward reference
    return Potato(color='red', size=10)

@app.get('/')
async def read_root(potato: Annotated[Potato, Depends(get_potato)]):
    return {'Hello': 'World'}
# OpenAPI schema doesn't include Potato definition correctly ❌

@dataclass
class Potato:
    color: str
    size: int

# ✓ WORKAROUND 1: Don't use __future__ annotations in route files
# Remove: from __future__ import annotations

# ✓ WORKAROUND 2: Use string literals for type hints
def get_potato() -> "Potato":
    return Potato(color='red', size=10)

# ✓ WORKAROUND 3: Define classes before they're used in dependencies
@dataclass
class Potato:
    color: str
    size: int

def get_potato() -> Potato:  # Now works ✓
    return Potato(color='red', size=10)

Issue #6: Pydantic v2 Path Parameter Union Type Breaking Change

Error : Path parameters with int | str always parse as str in Pydantic v2 Source : GitHub Issue #11251 | Community-sourced Why It Happens : Major breaking change when migrating from Pydantic v1 to v2. Union types with str in path/query parameters now always parse as str (worked correctly in v1).

Prevention :

from uuid import UUID

# ✗ PROBLEMATIC: Union with str in path parameter
@app.get("/int/{path}")
async def int_path(path: int | str):
    return str(type(path))
    # Pydantic v1: returns <class 'int'> for "123"
    # Pydantic v2: returns <class 'str'> for "123" ❌

@app.get("/uuid/{path}")
async def uuid_path(path: UUID | str):
    return str(type(path))
    # Pydantic v1: returns <class 'uuid.UUID'> for valid UUID
    # Pydantic v2: returns <class 'str'> ❌

# ✓ RIGHT: Avoid union types with str in path/query parameters
@app.get("/int/{path}")
async def int_path(path: int):
    return str(type(path))  # Works correctly ✓

# ✓ ALTERNATIVE: Use validators if type coercion needed
from pydantic import field_validator

class PathParams(BaseModel):
    path: int | str

    @field_validator('path')
    def coerce_to_int(cls, v):
        if isinstance(v, str) and v.isdigit():
            return int(v)
        return v

Issue #7: ValueError in field_validator Returns 500 Instead of 422

Error : 500 Internal Server Error when raising ValueError in custom validators Source : GitHub Discussion #10779 | Community-sourced Why It Happens : When raising ValueError inside a Pydantic @field_validator with Form fields, FastAPI returns 500 Internal Server Error instead of the expected 422 Unprocessable Entity validation error.

Prevention :

from typing import Annotated
from fastapi import Form
from pydantic import BaseModel, field_validator, ValidationError, Field

# ✗ WRONG: ValueError in validator
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValueError("Value must be positive")  # Returns 500! ❌
        return v

# ✓ RIGHT 1: Raise ValidationError instead
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValidationError("Value must be positive")  # Returns 422 ✓
        return v

# ✓ RIGHT 2: Use Pydantic's built-in constraints
class MyForm(BaseModel):
    value: Annotated[int, Field(gt=0)]  # Built-in validation, returns 422 ✓

Common Errors & Fixes

422 Unprocessable Entity

Cause : Request body doesn't match Pydantic schema

Debug :

Check /docs endpoint - test there first
Verify JSON structure matches schema
Check required vs optional fields

Fix : Add custom validation error handler:

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )

CORS Errors

Cause : Missing or misconfigured CORS middleware

Fix :

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Not "*" in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Async Blocking Event Loop

Cause : Blocking call in async route (e.g., time.sleep(), sync database client, CPU-bound operations)

Symptoms (production-scale):

Throughput plateaus far earlier than expected
Latency "balloons" as concurrency increases
Request pattern looks almost serial under load
Requests queue indefinitely when event loop is saturated
Small scattered blocking calls that aren't obvious (not infinite loops)

Fix : Use async alternatives:

# ✗ WRONG: Blocks event loop
import time
from sqlalchemy import create_engine  # Sync client

@app.get("/users")
async def get_users():
    time.sleep(0.1)  # Even small blocking adds up at scale!
    result = sync_db_client.query("SELECT * FROM users")  # Blocks!
    return result

# ✓ RIGHT 1: Use async database driver
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    await asyncio.sleep(0.1)  # Non-blocking
    result = await db.execute(select(User))
    return result.scalars().all()

# ✓ RIGHT 2: Use def (not async def) for CPU-bound routes
# FastAPI runs def routes in thread pool automatically
@app.get("/cpu-heavy")
def cpu_heavy_task():  # Note: def not async def
    return expensive_cpu_work()  # Runs in thread pool ✓

# ✓ RIGHT 3: Use run_in_executor for blocking calls in async routes
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@app.get("/mixed")
async def mixed_task():
    # Run blocking function in thread pool
    result = await asyncio.get_event_loop().run_in_executor(
        executor,
        blocking_function  # Your blocking function
    )
    return result

Sources : Production Case Study (Jan 2026) | Community-sourced

"Field required" for Optional Fields

Cause : Using Optional[str] without default

Fix :

# Wrong
description: Optional[str]  # Still required!

# Right
description: str | None = None  # Optional with default

Testing

# tests/test_main.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_root(client):
    response = await client.get("/")
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_create_item(client):
    response = await client.post(
        "/items",
        json={"name": "Test", "price": 9.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test"

Run: uv run pytest

Deployment

Uvicorn (Development)

uv run fastapi dev src/main.py

Uvicorn (Production)

uv run uvicorn src.main:app --host 0.0.0.0 --port 8000

Gunicorn + Uvicorn (Production with workers)

uv add gunicorn
uv run gunicorn src.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Docker

FROM python:3.12-slim

WORKDIR /app
COPY . .

RUN pip install uv && uv sync

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

References

Last verified : 2026-01-21 | Skill version : 1.1.0 | Changes : Added 7 known issues (form data bugs, background tasks, Pydantic v2 migration gotchas), expanded async blocking guidance with production patterns Maintainer : Jezweb | jeremy@jezweb.net

Weekly Installs

171

Repository

srbhr/resume-matcher

GitHub Stars

26.4K

First Seen

Jan 22, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykPass

Installed on

opencode130

gemini-cli128

codex128

github-copilot115

claude-code107

cursor101

agent-browser 浏览器自动化工具 - Vercel Labs 命令行网页操作与测试

157,400 周安装

FastAPI 生产级模板：Pydantic v2 + SQLAlchemy 异步 + JWT 身份验证

🇨🇳中文介绍

FastAPI Skill

快速开始

使用 uv 设置项目

最小工作示例

相关 Skills

项目结构（基于领域）

核心模式

Pydantic 模式（验证）

SQLAlchemy 模型（数据库）

数据库设置（异步 SQLAlchemy 2.0）

路由器模式

主应用

JWT 身份验证

身份验证模式

身份验证服务

身份验证依赖项

身份验证路由器

保护路由

配置

关键规则

务必做到

切勿做到

已知问题预防

问题 #1：表单数据丢失字段设置元数据

问题 #2：自定义响应静默覆盖 BackgroundTasks

问题 #3：可选表单字段在 TestClient 中中断（回归）

问题 #4：Pydantic Json 类型不适用于表单数据

问题 #5：带有前向引用的 Annotated 破坏 OpenAPI 生成

问题 #6：Pydantic v2 路径参数联合类型破坏性变更

问题 #7：field_validator 中的 ValueError 返回 500 而非 422

常见错误与修复

422 不可处理实体

CORS 错误

异步阻塞事件循环

可选字段出现 "Field required"

测试

部署

Uvicorn（开发）

Uvicorn（生产）

Gunicorn + Uvicorn（带工作进程的生产环境）

Docker

参考

🇺🇸English

FastAPI Skill

Quick Start

Project Setup with uv

Minimal Working Example

Project Structure (Domain-Based)

Core Patterns

Pydantic Schemas (Validation)

SQLAlchemy Models (Database)

Database Setup (Async SQLAlchemy 2.0)

Router Pattern

Main App

JWT Authentication

Auth Schemas

Auth Service

Auth Dependencies

Auth Router

Protect Routes

Configuration

Critical Rules

Always Do

Never Do

Known Issues Prevention

Issue #1: Form Data Loses Field Set Metadata

Issue #2: BackgroundTasks Silently Overwritten by Custom Response

Issue #3: Optional Form Fields Break with TestClient (Regression)

Issue #4: Pydantic Json Type Doesn't Work with Form Data

Issue #5: Annotated with ForwardRef Breaks OpenAPI Generation

Issue #6: Pydantic v2 Path Parameter Union Type Breaking Change

Issue #7: ValueError in field_validator Returns 500 Instead of 422

Common Errors & Fixes

422 Unprocessable Entity

CORS Errors

Async Blocking Event Loop

"Field required" for Optional Fields

Testing