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 路径
)

路由定义

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"],
    deprecated=False,
    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}

Cookie 和 Header

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}

参数验证

Path 验证

参数	含义
...	必需参数
gt	大于
ge	大于等于
lt	小于
le	小于等于

item_id: int = Path(..., gt=0, le=1000)

Query 验证

q: str | None = Query(
    None,
    min_length=3,
    max_length=50,
    regex="^[a-z]+$",
    title="搜索词",
    description="搜索关键词"
)

Body 验证

from fastapi import Body

item: Item = Body(
    ...,
    example={"name": "商品", "price": 99.9}
)

响应处理

响应模型

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

JSONResponse

from fastapi.responses import JSONResponse

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return JSONResponse(
        content={"item_id": item_id},
        status_code=200,
        headers={"X-Custom": "Value"}
    )

Pydantic 模型

基本模型

from pydantic import BaseModel, Field
from typing import Optional

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

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

注册路由器

from routers import items, users

app.include_router(items.router)
app.include_router(users.router, prefix="/api/v1")

依赖注入

基本依赖

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

类作为依赖

class CommonParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonParams = Depends()):
    return {"q": commons.q, "skip": commons.skip}

异常处理

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 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=["*"],
)

测试

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

常用状态码

状态码	含义	使用场景
200	OK	成功
201	Created	创建成功
204	No Content	删除成功
400	Bad Request	请求错误
401	Unauthorized	未认证
403	Forbidden	无权限
404	Not Found	未找到
422	Unprocessable Entity	验证错误
500	Internal Server Error	服务器错误

常用响应类型

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