关于 fastapi
fastapi 是面向 FastAPI 开发者的实践指南,涵盖 Pydantic v2、SQLAlchemy 2.0 异步数据库和 JWT 身份验证。它还整理了已知问题的规避方法与常见错误排查。
主要功能
- 01提供项目结构与路由组织模式
- 02使用 Pydantic v2 验证数据
- 03采用 SQLAlchemy 2.0 异步数据库
- 04实现 JWT 身份验证与路由保护
- 05说明已知问题及常见错误修复
适用场景
- 01搭建 FastAPI 项目基础结构
- 02实现带验证的异步数据库 API
- 03为 API 添加 JWT 登录与访问控制
- 04排查表单、OpenAPI 和异步阻塞问题
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ jezweb/ claude-skills --skill fastapi安装后在 Claude Code 中输入/fastapi确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
README.md.claude-plugin/plugin.jsontemplates/.env.examplepyproject.tomltemplates/src/__init__.pyconfig.pydatabase.pymain.pytemplates/src/auth/__init__.pydependencies.pymodels.pyrouter.pyschemas.pyservice.pytemplates/tests/__init__.pytest_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
关键规则
务必做到
- 将 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 请求体数据。
避免方法:
# ✗ 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 架构不匹配
调试:
- 检查
/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 错误
原因:缺少或配置错误的 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]