mcpField
F
fastapi

fastapi

github.com/jezweb开发1.1k2.0k 次安装3 周前更新
代码仓库

关于 fastapi

fastapi 是面向 FastAPI 开发者的实践指南,涵盖 Pydantic v2、SQLAlchemy 2.0 异步数据库和 JWT 身份验证。它还整理了已知问题的规避方法与常见错误排查。

主要功能

  1. 01提供项目结构与路由组织模式
  2. 02使用 Pydantic v2 验证数据
  3. 03采用 SQLAlchemy 2.0 异步数据库
  4. 04实现 JWT 身份验证与路由保护
  5. 05说明已知问题及常见错误修复

适用场景

  1. 01搭建 FastAPI 项目基础结构
  2. 02实现带验证的异步数据库 API
  3. 03为 API 添加 JWT 登录与访问控制
  4. 04排查表单、OpenAPI 和异步阻塞问题

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/jezweb/claude-skills --skill fastapi

安装后在 Claude Code 中输入/fastapi确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

README.md
.claude-plugin/
plugin.json
templates/
.env.example
pyproject.toml
templates/src/
__init__.py
config.py
database.py
main.py
templates/src/auth/
__init__.py
dependencies.py
models.py
router.py
schemas.py
service.py
templates/tests/
__init__.py
test_main.py

详细介绍

FastAPI 技能

经过生产环境验证的 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 的支持)

快速入门

使用 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
最小可运行示例
# 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

文档地址:http://127.0.0.1:8000/docs


项目结构(按领域划分)

为了便于维护,按领域而非文件类型组织:

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

核心模式

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() 设置验证约束
  • 分别定义 Create/Update/Response 模式
  • 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):
    # 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 身份验证

身份验证模式
# 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)
):
    # 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
保护路由
# 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)
    # ...

配置

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

创建 .env:

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

关键规则

务必做到
  1. 将 Pydantic 模式与 SQLAlchemy 模型分开 —— 用途不同,文件也应不同
  2. 对 I/O 操作使用异步方式 —— 数据库、HTTP 调用、文件访问
  3. 使用 Pydantic Field() 进行验证 —— 约束、默认值、说明
  4. 使用依赖注入 —— 使用 Depends() 注入数据库、身份验证和验证逻辑
  5. 返回适当的状态码 —— 创建时返回 201,删除时返回 204,等等
切勿这样做
  1. 切勿在异步路由中使用阻塞调用 —— 不要使用 time.sleep(),应使用 asyncio.sleep()
  2. 切勿将业务逻辑放在路由中 —— 使用服务层
  3. 切勿硬编码机密信息 —— 使用环境变量
  4. 切勿跳过验证 —— 务必使用 Pydantic 模式
  5. 生产环境中切勿在 CORS 来源中使用 * —— 指定确切的来源

避免已知问题

此技能可避免官方 FastAPI GitHub 和发行说明中记录的 7 个问题。

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

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

避免方法:

# ✗ 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 ✓
问题 #2:自定义 Response 会悄悄覆盖 BackgroundTasks

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

避免方法:

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

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

问题 #3:TestClient 会导致可选表单字段出错(回归问题)

错误:可选 Literal 字段出现 422: "Input should be 'abc' or 'def'" 错误 来源:GitHub Issue #12245 原因:从 FastAPI 0.114.0 开始,通过 TestClient 传入 None 时,带有 Literal 类型的可选表单字段无法通过验证。此问题在 0.113.0 中不存在。

避免方法:

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")
问题 #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

# ✗ 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 ✓
问题 #5:Annotated 与 ForwardRef 会导致 OpenAPI 生成失败

错误:依赖项类型的 OpenAPI 架构缺失或不正确 来源:GitHub Issue #13056 原因:使用 Annotated 和 Depends(),并搭配前向引用(来自 __future__ import annotations)时,OpenAPI 架构生成会失败或生成不正确的架构。

预防措施:

# ✗ 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)
问题 #6:Pydantic v2 路径参数联合类型的重大变更

错误:在 Pydantic v2 中,int | str 类型的路径参数始终会被解析为 str 来源:GitHub Issue #11251 | 来自社区 原因:从 Pydantic v1 迁移到 v2 时的一项重大破坏性变更。路径或查询参数中的联合类型包含 str 时,现在始终会被解析为 str(在 v1 中可以正常工作)。

预防措施:

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
问题 #7:field_validator 中的 ValueError 会返回 500,而不是 422

错误:在自定义验证器中引发 ValueError 时出现 500 Internal Server Error 来源:GitHub Discussion #10779 | 来自社区 原因:在包含表单字段的 Pydantic @field_validator 中引发 ValueError 时,FastAPI 会返回 500 Internal Server Error,而不是预期的 422 Unprocessable Entity 验证错误。

预防措施:

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 ✓

常见错误及修复方法

422 Unprocessable Entity

原因:请求正文与 Pydantic 架构不匹配

调试:

  1. 检查 /docs 端点——先在那里测试
  2. 确认 JSON 结构与架构匹配
  3. 检查必填字段和可选字段

修复方法:添加自定义验证错误处理器:

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 错误

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

修复方法:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Not "*" in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
异步操作阻塞事件循环

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

症状(生产环境规模):

  • 吞吐量远早于预期便达到上限
  • 并发量增加时,延迟会“大幅飙升”
  • 负载下的请求模式看起来几乎是串行的
  • 事件循环饱和时,请求会无限期排队
  • 存在不易察觉的零散阻塞调用(并非无限循环)

修复方法:使用异步替代方案:

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

来源:生产案例分析(2026 年 1 月) | 来自社区

可选字段出现“Field required”

原因:使用 Optional[str] 却未设置默认值

修复方法:

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

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

测试

# 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


部署

Uvicorn(开发环境)
uv run fastapi dev src/main.py
Uvicorn(生产环境)
uv run uvicorn src.main:app --host 0.0.0.0 --port 8000
Gunicorn + Uvicorn(使用 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"]

参考资料


最后验证时间:2026-01-21 | 技能版本:1.1.0 | 变更内容:新增 7 个已知问题(表单数据错误、后台任务、Pydantic v2 迁移注意事项),扩展了有关异步阻塞的指导并补充生产环境模式 维护者:Jezweb | [email protected]

评论

支持 Markdown