FastAPI Skill

FastAPI 实践指南

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)

经过生产环境验证的FastAPI实践模式，搭配Pydantic v2、异步SQLAlchemy 2.0和JWT认证。

最新版本（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的支持）

Quick Start

快速开始

Project Setup with uv

使用uv搭建项目

bash

undefined

bash

undefined

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

undefined

uv run fastapi dev src/main.py

undefined

Minimal Working Example

最小可用示例

python

undefined

python

undefined

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`

---

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`

文档地址：`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

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

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)

Pydantic 架构（验证）

python

undefined

python

undefined

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]`

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 Models (Database)

SQLAlchemy 模型（数据库）

python

undefined

python

undefined

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
)

undefined

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
)

undefined

Database Setup (Async SQLAlchemy 2.0)

数据库设置（异步SQLAlchemy 2.0）

python

undefined

python

undefined

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

undefined

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

undefined

Router Pattern

路由模式

python

undefined

python

undefined

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

undefined

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

undefined

Main App

主应用

python

undefined

python

undefined

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)

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)

---

app.include_router(auth_router) app.include_router(items_router)

---

JWT Authentication

JWT 认证

Auth Schemas

认证架构

python

undefined

python

undefined

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

undefined

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

undefined

Auth Service

认证服务

python

undefined

python

undefined

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

undefined

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

undefined

Auth Dependencies

认证依赖

python

undefined

python

undefined

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

undefined

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

undefined

Auth Router

认证路由

python

undefined

python

undefined

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

undefined

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

undefined

Protect Routes

保护路由

python

undefined

python

undefined

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) # ...

---

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

配置

python

undefined

python

undefined

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

---

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()


创建`.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.

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

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

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

Known Issues Prevention

已知问题预防

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

本指南可预防FastAPI官方GitHub和发布说明中记录的7个问题。

Issue #1: Form Data Loses Field Set Metadata

问题1：表单数据丢失字段集元数据

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:

python

undefined

错误：使用

Form()

时，

model.model_fields_set

包含默认值来源：GitHub Issue #13399 原因：表单数据解析会预加载默认值并传递给验证器，导致无法区分用户显式设置的字段和使用默认值的字段。此bug仅影响表单数据，不影响JSON请求体。

预防方案:

python

undefined

✗ 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! ❌

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 ✓

undefined

@app.post("/json") async def endpoint(model: MyModel): fields = model.model_fields_set # Works correctly ✓

undefined

Issue #2: BackgroundTasks Silently Overwritten by Custom Response

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

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:

python

undefined

错误：通过

BackgroundTasks

依赖添加的后台任务未运行来源：GitHub Issue #11215 原因：当返回带有

background

参数的自定义

Response

时，会覆盖所有注入到

BackgroundTasks

依赖中的任务。此行为未被文档记录，会导致静默失败。

预防方案:

python

undefined

✗ 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 )

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=...)`.

@app.get("/") async def endpoint(): return Response( content="Done", background=BackgroundTask(log_event) )


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

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

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

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:

python

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

错误：可选Literal字段返回

422: "Input should be 'abc' or 'def'"

来源：GitHub Issue #12245 原因：从FastAPI 0.114.0开始，当通过TestClient传递

None

时，带有

Literal

类型的可选表单字段验证失败。在0.113.0中可正常工作。

预防方案:

python

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 ❌

@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")

undefined

@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")

undefined

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

问题4：Pydantic Json类型无法与表单数据配合使用

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:

python

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

错误：

"JSON object must be str, bytes or bytearray"

来源：GitHub Issue #10997 原因：直接将Pydantic的

Json

类型与

Form()

一起使用会失败。必须将字段作为

str

接收并手动解析。

预防方案:

python

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 ✓

undefined

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 ✓

undefined

Issue #5: Annotated with ForwardRef Breaks OpenAPI Generation

问题5：带ForwardRef的Annotated会破坏OpenAPI生成

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:

python

undefined

错误：依赖类型的OpenAPI架构缺失或不正确来源：GitHub Issue #13056 原因：当使用

Annotated

搭配

Depends()

和前向引用（来自

__future__ import annotations

）时，OpenAPI架构生成会失败或产生不正确的架构。

预防方案:

python

undefined

✗ 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'}

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)

undefined

@dataclass class Potato: color: str size: int

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

undefined

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

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

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:

python

from uuid import UUID

错误：Pydantic v2中，带有

int | str

的路径参数始终被解析为

str

来源：GitHub Issue #11251 | 社区反馈原因：从Pydantic v1迁移到v2时的重大破坏性变更。路径/查询参数中带有

str

的联合类型现在始终被解析为

str

（在v1中可正常工作）。

预防方案:

python

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'> ❌

@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

undefined

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

undefined

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

问题7：field_validator中的ValueError返回500而非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:

python

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

错误：自定义验证器中抛出

ValueError

时返回500内部服务器错误来源：GitHub Discussion #10779 | 社区反馈原因：当在Pydantic

@field_validator

中针对表单字段抛出

ValueError

时，FastAPI返回500内部服务器错误而非预期的422无法处理的实体验证错误。

预防方案:

python

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

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

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 ✓

---

class MyForm(BaseModel): value: Annotated[int, Field(gt=0)] # Built-in validation, returns 422 ✓

---

Common Errors & Fixes

常见错误与修复

422 Unprocessable Entity

422 无法处理的实体

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:

python

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}
    )

原因：请求体不符合Pydantic架构

调试:

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

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

python

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

CORS 错误

Cause: Missing or misconfigured CORS middleware

Fix:

python

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

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

修复:

python

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 生产环境不要用"*"
    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:

python

undefined

原因：异步路由中使用阻塞调用（如

time.sleep()

、同步数据库客户端、CPU密集型操作）

生产环境症状:

吞吐量远早于预期达到瓶颈
并发增加时延迟急剧上升
负载下请求模式几乎是串行的
事件循环饱和时请求无限排队
不明显的小型分散阻塞调用（非无限循环）

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

python

undefined

✗ 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

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()

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)](https://www.techbuddies.io/2026/01/10/case-study-fixing-fastapi-event-loop-blocking-in-a-high-traffic-api/) | Community-sourced

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


**来源**：[生产环境案例研究（2026年1月）](https://www.techbuddies.io/2026/01/10/case-study-fixing-fastapi-event-loop-blocking-in-a-high-traffic-api/) | 社区反馈

"Field required" for Optional Fields

可选字段提示“Field required”

Cause: Using

Optional[str]

without default

Fix:

python

undefined

原因：使用

Optional[str]

但未设置默认值

修复:

python

undefined

Wrong

description: Optional[str] # Still required!

Right

description: str | None = None # Optional with default

---

description: str | None = None # Optional with default

---

Testing

测试

python

undefined

python

undefined

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`

---

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`

---

Deployment

部署

Uvicorn (Development)

Uvicorn（开发环境）

bash

uv run fastapi dev src/main.py

bash

uv run fastapi dev src/main.py

Uvicorn (Production)

Uvicorn（生产环境）

bash

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

bash

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

Gunicorn + Uvicorn (Production with workers)

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

bash

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

bash

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

Docker

dockerfile

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"]

dockerfile

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

最后验证时间：2026-01-21 | 指南版本：1.1.0 | 更新内容：新增7个已知问题（表单数据bug、后台任务、Pydantic v2迁移陷阱），扩展了异步阻塞的生产环境模式指导 维护者：Jezweb | jeremy@jezweb.net

fastapi

Original

Translation

FastAPI Skill

FastAPI 实践指南

Quick Start

快速开始

Project Setup with uv

使用uv搭建项目

Create project

Create project

Add dependencies

Add dependencies

Run development server

Run development server

Minimal Working Example

最小可用示例

src/main.py

src/main.py

Project Structure (Domain-Based)

基于领域的项目结构

Core Patterns

核心模式

Pydantic Schemas (Validation)

Pydantic 架构（验证）

src/items/schemas.py

src/items/schemas.py

SQLAlchemy Models (Database)

SQLAlchemy 模型（数据库）

src/items/models.py

src/items/models.py

Database Setup (Async SQLAlchemy 2.0)

数据库设置（异步SQLAlchemy 2.0）

src/database.py

src/database.py

Router Pattern

路由模式

src/items/router.py

src/items/router.py

Main App

主应用

src/main.py

src/main.py

CORS middleware

CORS middleware

Include routers

Include routers

JWT Authentication

JWT 认证

Auth Schemas

认证架构

src/auth/schemas.py

src/auth/schemas.py

Auth Service

认证服务

src/auth/service.py

src/auth/service.py

Auth Dependencies

认证依赖

src/auth/dependencies.py

src/auth/dependencies.py

Auth Router

认证路由

src/auth/router.py

src/auth/router.py

Protect Routes

保护路由

In any router

In any router

Configuration

配置

src/config.py

src/config.py

Critical Rules

重要规则

Always Do

必须遵守

Never Do

绝对禁止

Known Issues Prevention