mcpField
F
fastapi

fastapi

github.com/srbhr开发29.2k179 次安装4 天前更新
代码仓库

关于 fastapi

面向 FastAPI 开发者的编码技能,涵盖 Pydantic v2 与异步 SQLAlchemy 2.0。提供项目结构、实现模式及常见问题规避建议,适用于 API 开发与维护。

主要功能

  1. 01Pydantic 模式与校验方法
  2. 02异步 SQLAlchemy 数据库模式
  3. 03JWT 身份验证示例
  4. 04按领域组织项目结构
  5. 05常见问题规避指南

适用场景

  1. 01启动 FastAPI API 项目
  2. 02规划按领域划分的代码结构
  3. 03添加 JWT 保护路由
  4. 04规避常见校验与异步错误

安装

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

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

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

详细介绍

FastAPI Skill

经过生产环境验证的 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 origins 中使用 *:应指定确切的 origins

已知问题预防

本 Skill 可预防 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: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 开始,带 Literal 类型的可选表单字段在通过 TestClient 传入 None 时校验失败。在 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 | 社区提供 原因:在带有 Form 字段的 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