FastAPI 速查表

本页面汇总了 FastAPI 开发中最常用的语法和知识点，方便快速查阅。

应用创建

from fastapi import FastAPI

app = FastAPI(
    title="API 标题",
    description="API 描述",
    version="1.0.0",
    docs_url="/docs",       # Swagger UI 路径
    redoc_url="/redoc",     # ReDoc 路径
    root_path="/api/v1",    # 代理路径前缀（可选）
)

路由定义

HTTP 方法

@app.get("/")           # 获取资源
@app.post("/")          # 创建资源
@app.put("/{id}")       # 完整更新
@app.patch("/{id}")     # 部分更新
@app.delete("/{id}")    # 删除资源

路由装饰器选项

@app.get(
    "/items/{item_id}",
    summary="简短描述",
    description="详细描述",
    response_model=Item,
    status_code=200,
    tags=["items"],
    responses={
        404: {"description": "未找到"},
        400: {"description": "错误请求"}
    }
)

参数类型

路径参数

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 带验证
@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., ge=1, le=1000, title="商品ID")
):
    return {"item_id": item_id}

查询参数

@app.get("/items/")
async def read_items(
    skip: int = 0,
    limit: int = 10,
    q: str | None = None
):
    return {"skip": skip, "limit": limit, "q": q}

# 带验证
@app.get("/items/")
async def read_items(
    q: str | None = Query(None, min_length=3, max_length=50)
):
    return {"q": q}

请求体

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    description: str | None = None

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

表单数据

from fastapi import Form

@app.post("/login/")
async def login(
    username: str = Form(...),
    password: str = Form(...)
):
    return {"username": username}

文件上传

from fastapi import UploadFile, File

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {"filename": file.filename}

from fastapi import Cookie, Header

@app.get("/items/")
async def read_items(
    session_id: str | None = Cookie(None),
    user_agent: str | None = Header(None)
):
    return {"session_id": session_id, "user_agent": user_agent}

参数验证

数值验证

参数	含义
`gt`	大于
`ge`	大于等于
`lt`	小于
`le`	小于等于
`multiple_of`	是某数的倍数

item_id: int = Path(..., gt=0, le=1000)
price: float = Field(..., gt=0, description="价格必须大于0")

字符串验证

q: str | None = Query(
    None,
    min_length=3,
    max_length=50,
    pattern="^[a-z]+$"
)

响应处理

响应模型

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    return {"id": item_id, "name": "Item", "price": 99.9}

# 响应模型列表
@app.get("/items/", response_model=list[Item])
async def list_items():
    return items

# 排除字段
@app.get("/users/{user_id}", response_model=User, response_model_exclude={"password"})
async def read_user(user_id: int):
    return user

响应状态码

from fastapi import status

@app.post("/items/", status_code=201)
# 或
@app.post("/items/", status_code=status.HTTP_201_CREATED)

响应类

from fastapi.responses import (
    JSONResponse,        # JSON 响应
    HTMLResponse,        # HTML 响应
    PlainTextResponse,   # 纯文本
    RedirectResponse,    # 重定向
    StreamingResponse,   # 流式响应
    FileResponse,        # 文件响应
)

Pydantic 模型

基本模型

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str
    price: float = Field(..., gt=0, description="价格必须大于0")
    description: str | None = None
    tags: list[str] = []

    model_config = {
        "json_schema_extra": {
            "example": {
                "name": "商品名",
                "price": 99.9,
                "description": "商品描述"
            }
        }
    }

模型继承

class ItemBase(BaseModel):
    name: str
    description: str | None = None

class ItemCreate(ItemBase):
    price: float

class Item(ItemBase):
    id: int
    price: float

路由器

from fastapi import APIRouter

router = APIRouter(
    prefix="/items",
    tags=["items"],
    responses={404: {"description": "Not found"}}
)

@router.get("/")
async def list_items():
    return []

@router.get("/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 注册路由器
app.include_router(router)

依赖注入

基本依赖

from fastapi import Depends

async def common_params(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_params)):
    return commons

类型别名

from typing import Annotated

CommonsDep = Annotated[dict, Depends(common_params)]

@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons

Lifespan 事件

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    print("应用启动")
    yield
    # 关闭时执行
    print("应用关闭")

app = FastAPI(lifespan=lifespan)

流式 JSON Lines（0.134.0+）

from collections.abc import AsyncIterable
from pydantic import BaseModel

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

# 使用 yield 流式返回数据
@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item

# 同步生成器
@app.get("/items/stream-sync")
def stream_items_sync() -> Iterable[Item]:
    for item in items:
        yield item

OpenAPI Webhooks

from pydantic import BaseModel

class Subscription(BaseModel):
    username: str
    fee: float

# 定义 Webhook（仅用于文档）
@app.webhooks.post("new-subscription")
def new_subscription(body: Subscription):
    """用户订阅时发送此事件"""
    pass

# 业务路由
@app.post("/subscribe")
async def subscribe(user: Subscription):
    # 实际发送 Webhook 需要自己实现
    await send_webhook("new-subscription", user)
    return {"status": "subscribed"}

子应用与挂载

# 创建子应用
sub_app = FastAPI(title="子应用")

@sub_app.get("/")
async def sub_root():
    return {"message": "子应用"}

# 挂载到主应用
app.mount("/subapi", sub_app)

# 子应用访问：/subapi/
# 子应用文档：/subapi/docs

中间件

自定义中间件

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

CORS 中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

异常处理

HTTPException

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "Not Found"}
        )
    return items[item_id]

自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something."}
    )

测试

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

运行命令

# 开发模式（自动重载）
fastapi dev main.py

# 生产模式
fastapi run main.py

# Uvicorn 命令
uvicorn main:app --reload
uvicorn main:app --host 0.0.0.0 --port 8000
uvicorn main:app --workers 4

# 代理配置
fastapi run --forwarded-allow-ips="*"
fastapi run --root-path /api/v1

常用状态码

状态码	常量	含义	使用场景
200	`HTTP_200_OK`	成功	默认状态码
201	`HTTP_201_CREATED`	已创建	POST 创建成功
204	`HTTP_204_NO_CONTENT`	无内容	DELETE 成功
400	`HTTP_400_BAD_REQUEST`	错误请求	参数错误
401	`HTTP_401_UNAUTHORIZED`	未认证	需要登录
403	`HTTP_403_FORBIDDEN`	禁止访问	无权限
404	`HTTP_404_NOT_FOUND`	未找到	资源不存在
422	`HTTP_422_UNPROCESSABLE_ENTITY`	无法处理	数据验证失败
500	`HTTP_500_INTERNAL_SERVER_ERROR`	服务器错误	内部错误

Server-Sent Events（SSE）

from collections.abc import AsyncIterable
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()

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

# 简单 SSE：yield Pydantic 模型或字典
@app.get("/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item  # 自动 JSON 编码

# 完整控制：使用 ServerSentEvent
@app.get("/stream-full", response_class=EventSourceResponse)
async def stream_full() -> AsyncIterable[ServerSentEvent]:
    for i, item in enumerate(items):
        yield ServerSentEvent(
            data=item,           # 数据负载
            event="item_update", # 事件类型
            id=str(i),           # 事件 ID（支持断点续传）
            retry=5000           # 重连间隔（毫秒）
        )

# POST 请求的 SSE
@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    for token in generate_tokens(prompt.text):
        yield ServerSentEvent(data={"token": token}, event="token")
    yield ServerSentEvent(raw_data="[DONE]", event="done")

SSE 字段说明：

字段	类型	说明
`data`	Any	事件数据，自动 JSON 序列化
`raw_data`	str	原始字符串，不进行 JSON 编码
`event`	str	事件类型名称
`id`	str	事件 ID，用于断点续传
`retry`	int	重连间隔（毫秒）
`comment`	str	注释，不发送给客户端

流式 JSON Lines

from collections.abc import AsyncIterable, Iterable
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

# 异步生成器（推荐）
@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item  # 每行一个 JSON

# 同步生成器
@app.get("/items/stream-sync")
def stream_items_sync() -> Iterable[Item]:
    for item in items:
        yield item

# 无限流
@app.get("/sensors/stream")
async def stream_sensors() -> AsyncIterable[SensorData]:
    while True:
        yield SensorData(temperature=get_temp())
        await asyncio.sleep(1)

JSON Lines 格式：

{"name": "商品A", "price": 99.9}
{"name": "商品B", "price": 199.9}

OpenAPI Webhooks

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Subscription(BaseModel):
    username: str
    fee: float

# 定义 Webhook（仅用于文档）
@app.webhooks.post("subscription.created")
def subscription_created(body: Subscription):
    """用户订阅时发送此事件"""
    pass

@app.webhooks.post("subscription.cancelled")
def subscription_cancelled(body: SubscriptionCancelled):
    """取消订阅时发送此事件"""
    pass

# 正常的 API 路由
@app.post("/subscribe")
async def subscribe(user: UserCreate):
    # ... 业务逻辑
    # 发送 Webhook 需要自己实现
    await send_webhook("subscription.created", subscription)
    return {"status": "success"}

Pydantic Settings 配置

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str
    secret_key: str
    debug: bool = False
    
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False
    )

@lru_cache
def get_settings() -> Settings:
    return Settings()

# 使用依赖注入
SettingsDep = Annotated[Settings, Depends(get_settings)]

@app.get("/info")
async def info(settings: SettingsDep):
    return {"app_name": settings.app_name}

常用命令

# 开发模式（自动重载）
fastapi dev main.py

# 生产模式
fastapi run main.py --workers 4

# 代理配置
fastapi run --forwarded-allow-ips="*"
fastapi run --root-path /api/v1

# Uvicorn 命令
uvicorn main:app --reload
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

客户端生成

# 使用 openapi-typescript-codegen
npx openapi-typescript-codegen --input http://localhost:8000/openapi.json --output ./client

# 使用 openapi-generator
openapi-generator-cli generate -i http://localhost:8000/openapi.json -g python -o ./client

应用创建​

路由定义​

HTTP 方法​

路由装饰器选项​

参数类型​

路径参数​

查询参数​

请求体​

表单数据​

文件上传​

Cookie 和 Header​

参数验证​

数值验证​

字符串验证​

响应处理​

响应模型​

响应状态码​

响应类​

Pydantic 模型​

基本模型​

模型继承​

路由器​

依赖注入​

基本依赖​

类型别名​

Lifespan 事件​

流式 JSON Lines（0.134.0+）​

OpenAPI Webhooks​

子应用与挂载​

中间件​

自定义中间件​

CORS 中间件​

异常处理​

HTTPException​

自定义异常处理器​

测试​

运行命令​

常用状态码​

Server-Sent Events（SSE）​

流式 JSON Lines​

OpenAPI Webhooks​

Pydantic Settings 配置​

常用命令​

客户端生成​

应用创建

路由定义

HTTP 方法

路由装饰器选项

参数类型

路径参数

查询参数

请求体

表单数据

文件上传

Cookie 和 Header

参数验证

数值验证

字符串验证

响应处理

响应模型

响应状态码

响应类

Pydantic 模型

基本模型

模型继承

路由器

依赖注入

基本依赖

类型别名

Lifespan 事件

流式 JSON Lines（0.134.0+）

OpenAPI Webhooks

子应用与挂载

中间件

自定义中间件

CORS 中间件

异常处理

HTTPException

自定义异常处理器

测试

运行命令

常用状态码

Server-Sent Events（SSE）

流式 JSON Lines

OpenAPI Webhooks

Pydantic Settings 配置

常用命令

客户端生成