入门
第一个 API
FastAPI 基于 Starlette 和 Pydantic。uvicorn 是 ASGI 服务器。--reload 用于在开发期间自动重载。
# install FastAPI and uvicorn
pip install fastapi uvicorn
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
# run: uvicorn main:app --reload使用 Uvicorn 运行
应用实例 'main:app' 表示模块 'main' 中的变量 'app'。--reload 仅用于开发；生产环境中不要使用，应改用进程管理器。
# development with auto-reload
uvicorn main:app --reload
# specify host and port
uvicorn main:app --host 0.0.0.0 --port 8000
# run programmatically
import uvicorn
if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)异步路径操作
路径操作可以用 'def'（在线程池中运行）或 'async def' 声明。当函数通过异步库执行 I/O 时使用 async def；在 async def 中混用阻塞调用会阻塞事件循环。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "hello"}
@app.get("/sync")
def read_sync():
return {"message": "sync also works"}
# use async when calling async libs (httpx, databases)
@app.get("/data")
async def fetch_data():
import httpx
async with httpx.AsyncClient() as client:
r = await client.get("https://api.example.com")
return r.json()路径操作装饰器
FastAPI 支持所有 HTTP 方法：GET、POST、PUT、PATCH、DELETE、OPTIONS、HEAD。每个方法对应一个装饰器。在装饰器上使用 status_code= 可覆盖默认的成功状态码。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items")
def list_items(): ...
@app.post("/items")
def create_item(): ...
@app.put("/items/{item_id}")
def replace_item(item_id: int): ...
@app.patch("/items/{item_id}")
def update_item(item_id: int): ...
@app.delete("/items/{item_id}")
def delete_item(item_id: int): ...
@app.options("/items")
def options_items(): ...
@app.head("/items")
def head_items(): ...JSON 响应与状态码
返回 dict/list 会自动转换为 JSON，状态码为 200（或你指定的 status_code）。若需完全控制状态码、头部或内容，请直接返回 JSONResponse 或 Response。
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/plain")
def plain():
return {"key": "value"} # auto JSON, 200 OK
@app.post("/created", status_code=status.HTTP_201_CREATED)
def created():
return {"id": 1}
@app.get("/custom")
def custom():
return JSONResponse(
status_code=418,
content={"error": "I'm a teapot"},
headers={"X-Custom": "yes"},
)项目结构
使用 APIRouter 将不断增长的应用拆分为模块，然后通过 app.include_router() 注册。将 Pydantic 模型、数据库设置和依赖项放在单独的文件中以便维护。
myapp/
main.py # creates the FastAPI app
routers/
users.py # APIRouter for /users
items.py # APIRouter for /items
models/
schemas.py # Pydantic models
database.py # DB engine & session
dependencies.py # shared Depends functions
tests/
test_main.py
# main.py
from fastapi import FastAPI
from routers import users, items
app = FastAPI()
app.include_router(users.router)
app.include_router(items.router, prefix="/items")路径参数
基本路径参数
从路径中捕获的值默认是字符串，除非添加类型注解。路径中的参数名必须与函数参数名完全匹配。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id):
return {"item_id": item_id}
# GET /items/42 -> {"item_id": "42"} (string by default)类型转换
添加类型注解（int、float、bool、str、Enum）后，FastAPI 会验证并转换路径值。无效值返回 422 响应和清晰的错误信息。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id}
# GET /items/3 -> {"item_id": 3}
# GET /items/abc -> 422 Validation Error (not an integer)路径验证
Path() 为路径参数添加元数据和约束。数值约束：ge (>=)、gt (>)、le (<=)、lt (<)。路径参数总是必填的，因此 Path() 不能设置默认值。
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(
item_id: int = Path(
title="The ID of the item",
description="Must be a positive integer",
ge=1,
le=1000,
),
):
return {"item_id": item_id}数值约束
对 int 或 float 路径参数使用 ge/gt/le/lt 来强制数值范围。组合约束可提供精确的边界验证并自动返回 422 错误。
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/ratio/{value}")
def ratio(
value: float = Path(gt=0, lt=1), # 0 < value < 1
):
return {"value": value}
@app.get("/page/{page}")
def page(
page: int = Path(ge=1), # >= 1
size: int = Path(le=100), # <= 100
):
return {"page": page, "size": size}路由顺序很重要
路由按声明顺序匹配。在动态模式（如 /users/{user_id}）之前声明具体路径（如 /users/me），否则动态路由会遮蔽具体路由。
from fastapi import FastAPI
app = FastAPI()
@app.get("/users/me")
def read_current_user():
return {"user": "me"}
@app.get("/users/{user_id}")
def read_user(user_id: str):
return {"user_id": user_id}
# /users/me MUST be declared before /users/{user_id}
# otherwise 'me' would match the {user_id} pattern枚举路径参数
使用 Enum 类型作为路径参数可将其限制为预定义值，并在 OpenAPI 文档中生成枚举 schema。继承 (str, Enum) 使值序列化为字符串。
from enum import Enum
from fastapi import FastAPI
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
app = FastAPI()
@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
if model_name is ModelName.alexnet:
return {"model": model_name, "layers": 5}
return {"model": model_name}
# GET /models/resnet -> {"model": "resnet"}查询参数
基本查询参数
不在路径中的函数参数会成为查询参数。提供默认值使其变为可选；当查询键缺失时使用默认值。
from fastapi import FastAPI
app = FastAPI()
fake_items = [{"item": "a"}, {"item": "b"}, {"item": "c"}]
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
return fake_items[skip : skip + limit]
# GET /items?skip=0&limit=2
# GET /items -> skip=0, limit=10可选查询参数
使用 Optional[str] = None（或 Python 3.10+ 的 str | None = None）表示可省略的查询参数。必填参数只需不设默认值。
from typing import Optional
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: str, q: Optional[str] = None):
if q:
return {"item_id": item_id, "q": q}
return {"item_id": item_id}
# GET /items/1?q=hello
# GET /items/1 (q is None)查询验证
Query() 添加验证：字符串用 min_length/max_length，数字用 ge/gt/le/lt，字符串格式用 pattern（正则）。使用 default= 设置值的同时应用约束。
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/search")
def search(q: str = Query(min_length=3, max_length=50, pattern="^[a-zA-Z]+$")):
return {"q": q}
@app.get("/page")
def page(limit: int = Query(default=10, ge=1, le=100)):
return {"limit": limit}布尔值转换
布尔查询参数接受多种真值：true、1、yes、on（不区分大小写）。其他值被解释为 false。无效类型如 ?active=maybe 返回 422。
from fastapi import FastAPI
app = FastAPI()
@app.get("/flags")
def flags(active: bool = False):
return {"active": active}
# GET /flags?active=true -> {"active": true}
# GET /flags?active=1 -> {"active": true}
# GET /flags?active=yes -> {"active": true}
# GET /flags?active=off -> {"active": false}列表查询参数
List[str] 参数接受查询字符串中重复的相同键。使用 Query(default=[]) 默认为空列表，或 Query(default=None) 允许缺失值。
from typing import List
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items")
def list_items(q: List[str] = Query(default=[])):
return {"q": q}
# GET /items?q=a&q=b&q=c -> {"q": ["a", "b", "c"]}
# declare default=None to accept a list or null必填查询参数
使用 Query(...)（省略号）声明一个仍带验证元数据的必填查询参数。没有默认值且没有 Query(...) 的普通注解参数也是必填的。
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items")
def read_item(q: str = Query(...)):
return {"q": q}
# GET /items?q=hello -> ok
# GET /items -> 422 (q is required)
# Query(...) (Ellipsis) marks a parameter as required请求体（Pydantic）
BaseModel 基础
声明为函数参数的 Pydantic BaseModel 成为 JSON 请求体。FastAPI 验证载荷、转换类型，无效数据返回 422。带默认值的字段是可选的。
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
in_stock: bool = True
app = FastAPI()
@app.post("/items")
def create_item(item: Item):
return item
# POST /items body: {"name": "Apple", "price": 0.5}
# -> {"name": "Apple", "price": 0.5, "in_stock": true}字段类型与约束
Field() 为模型属性添加约束和元数据：字符串用 min_length/max_length，数字用 gt/ge/lt/le，可变默认值用 default_factory。model_config 的 json_schema_extra 为文档添加示例。
from pydantic import BaseModel, Field
from typing import Optional
class Item(BaseModel):
name: str = Field(min_length=1, max_length=100)
price: float = Field(gt=0, description="must be positive")
tax: Optional[float] = None
tags: list[str] = Field(default_factory=list)
model_config = {
"json_schema_extra": {
"examples": [{"name": "Apple", "price": 0.5, "tags": ["fruit"]}]
}
}嵌套模型
模型可以嵌套：将另一个 BaseModel 声明为字段类型。FastAPI 递归验证整个嵌套结构，生成的 OpenAPI schema 也反映嵌套对象。
from pydantic import BaseModel
from typing import Optional
class Image(BaseModel):
url: str
name: str
class Item(BaseModel):
name: str
description: Optional[str] = None
image: Optional[Image] = None
images: list[Image] = []
# valid body:
# {"name": "Phone", "image": {"url": "http://x/a.png", "name": "a"}}字段验证
使用 @field_validator（Pydantic v2）为每个字段添加自定义验证逻辑。抛出 ValueError 可拒绝输入（FastAPI 将其转为 422）。返回（可能转换后的）值以保留它。
from pydantic import BaseModel, field_validator
class User(BaseModel):
name: str
email: str
@field_validator("email")
@classmethod
def email_must_contain_at(cls, v: str) -> str:
if "@" not in v:
raise ValueError("must contain @")
return v.lower()
@field_validator("name")
@classmethod
def name_stripped(cls, v: str) -> str:
return v.strip()可选字段与默认值
只有没有默认值且不是 Optional 的字段才是必填的。Optional[str] = None 是可选可空字段的标准模式；像 0.0 这样的普通默认值使字段可选并使用该值。
from pydantic import BaseModel
from typing import Optional
class Item(BaseModel):
name: str # required
description: Optional[str] = None # optional, defaults to None
price: float = 0.0 # optional, defaults to 0.0
tags: list[str] = [] # optional, defaults to empty list
# All of these are valid:
# {"name": "X"}
# {"name": "X", "price": 9.99, "tags": ["a"]}Body 与 Path、Query 共存
路径、查询和请求体参数可在同一操作中共存。FastAPI 判断依据：在路径中为路径参数，Pydantic 模型（或 Body()）为请求体，否则为查询。可选请求体需设默认值为 None。
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
app = FastAPI()
@app.put("/items/{item_id}")
def update_item(
item_id: int = Path(ge=1),
q: Optional[str] = Query(default=None),
item: Item = None,
):
result = {"item_id": item_id, **item.model_dump()}
if q:
result["q"] = q
return result响应模型
response_model 基础
response_model 将返回数据过滤为仅模型中定义的字段，即使返回的对象有额外字段。这是隐藏密码等敏感字段的标准方式。
from fastapi import FastAPI
from pydantic import BaseModel
class User(BaseModel):
username: str
email: str
hashed_password: str
class UserOut(BaseModel):
username: str
email: str
app = FastAPI()
@app.get("/users/me", response_model=UserOut)
def read_user():
return User(username="bob", email="[email protected]", hashed_password="secret")
# response: {"username": "bob", "email": "[email protected]"} (no password)排除字段
response_model_include 和 response_model_exclude 让你按路由白名单或黑名单字段，无需定义新模型。传入字段名集合。它们在 response_model 之上额外应用。
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
secret: str
app = FastAPI()
@app.get("/items/{id}", response_model=Item, response_model_exclude={"secret"})
def read_item(id: int):
return {"name": "Apple", "price": 0.5, "secret": "hidden"}
@app.get("/items", response_model=list[Item], response_model_include={"name", "price"})
def list_items():
return [{"name": "Apple", "price": 0.5, "secret": "x"}]列表响应
使用 response_model=list[Model]（或 List[Model]）声明端点返回对象数组。FastAPI 会验证并序列化列表中的每个元素。
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
app = FastAPI()
@app.get("/items", response_model=list[Item])
def list_items():
return [
{"name": "Apple", "price": 0.5},
{"name": "Banana", "price": 0.3},
]
# response: [{"name": "Apple", "price": 0.5}, {"name": "Banana", "price": 0.3}]排除未设置/默认值
exclude_unset 省略客户端从未发送的字段（只显示显式设置的）。exclude_defaults 省略等于默认值的字段；exclude_none 省略为 None 的字段。适用于稀疏响应。
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str | None = None
price: float = 0.0
tax: float = 0.0
app = FastAPI()
@app.get("/items/{id}", response_model=Item, response_model_exclude_unset=True)
def read_item(id: int):
return {"name": "Apple", "price": 0.5}
# -> {"name": "Apple", "price": 0.5} (description, tax omitted)
# also: response_model_exclude_defaults=True
# response_model_exclude_none=True按别名响应
Field(alias=...) 让传入 JSON 使用与 Python 属性不同的键名。response_model_by_alias=True 输出时也使用别名。populate_by_name=True 还允许输入时使用原始属性名。
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field
class Item(BaseModel):
model_config = ConfigDict(populate_by_name=True)
name: str = Field(alias="itemName")
price: float
app = FastAPI()
@app.post("/items", response_model=Item, response_model_by_alias=True)
def create(item: Item):
return item
# input: {"itemName": "Apple", "price": 0.5}
# output: {"itemName": "Apple", "price": 0.5}纯字典响应
返回 dict/list 会自动序列化为 JSON。使用 response_class 更改默认响应类型（HTMLResponse、PlainTextResponse），或直接返回 Response 子类以获得完全控制。
from fastapi import FastAPI
from fastapi.responses import JSONResponse, PlainTextResponse
app = FastAPI()
@app.get("/dict")
def dict_resp():
return {"a": 1, "b": 2} # auto JSON
@app.get("/text", response_class=PlainTextResponse)
def text_resp():
return "plain text response"
@app.get("/raw")
def raw_resp():
return JSONResponse(content={"a": 1}, status_code=200)表单与文件
表单数据
Form() 声明从 application/x-www-form-urlencoded 或 multipart 表单数据解析的参数。不能在同一操作中将 Form() 与 JSON 请求体（BaseModel）混用——只能选一种内容类型。
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login")
def login(username: str = Form(), password: str = Form()):
return {"username": username}
# Content-Type: application/x-www-form-urlencoded
# body: username=bob&password=secretUploadFile 基础
UploadFile 是流式文件表示，具有 .filename、.content_type、.read()、.write()、.close()。在 async def 中使用 'await file.read()'。大文件建议使用 file.file（SpooledTemporaryFile）避免全部加载到内存。
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {
"filename": file.filename,
"size": len(contents),
"content_type": file.content_type,
}多个文件
声明 list[UploadFile] 可在一个请求中接受多个文件。每个文件以相同字段名发送。列表中文件的顺序与发送顺序一致。
from fastapi import FastAPI, UploadFile
app = FastAPI()
@app.post("/uploads")
async def upload_many(files: list[UploadFile]):
return [
{"filename": f.filename, "size": len(await f.read())}
for f in files
]
# curl: curl -F "[email protected]" -F "[email protected]" http://localhost:8000/uploads文件与表单同时使用
Form() 和 UploadFile（File()）可在同一端点组合使用，因为两者都使用 multipart/form-data。但仍不能将它们与 JSON BaseModel 请求体组合。
from fastapi import FastAPI, UploadFile, File, Form
app = FastAPI()
@app.post("/items")
async def create_item(
name: str = Form(),
description: str = Form(),
file: UploadFile = File(),
):
return {
"name": name,
"description": description,
"file": file.filename,
}保存上传文件
对于小文件，shutil.copyfileobj 高效地将 SpooledTemporaryFile 流式传输到磁盘。对于超大文件，分块读取（如 1MB）以降低内存。务必清理 file.filename 以防止路径遍历攻击。
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile
app = FastAPI()
@app.post("/save")
async def save_file(file: UploadFile):
dest = Path("uploads") / file.filename
dest.parent.mkdir(exist_ok=True)
with dest.open("wb") as buf:
shutil.copyfileobj(file.file, buf)
return {"saved_to": str(dest)}
# async alternative for large files:
# async with dest.open("wb") as buf:
# while chunk := await file.read(1024 * 1024):
# buf.write(chunk)文件元数据与大小
在处理前验证 file.content_type 和大小。read() 后光标在末尾；如需再次读取，调用 'await file.seek(0)' 回退。真正的类型检查应检查文件魔数，而不仅仅是头部。
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/validate")
async def validate(file: UploadFile = File(...)):
# allowed types
allowed = {"image/png", "image/jpeg"}
if file.content_type not in allowed:
return {"error": "unsupported type"}
contents = await file.read()
max_size = 5 * 1024 * 1024 # 5 MB
if len(contents) > max_size:
return {"error": "file too large"}
await file.seek(0) # rewind so later reads work
return {"filename": file.filename, "size": len(contents)}依赖注入
基本 Depends
Depends() 将函数的返回值注入到路径操作中。共享逻辑（分页、解析、数据库会话）集中在一处并在多个端点复用。依赖项的参数也会在 OpenAPI 中记录。
from fastapi import FastAPI, Depends
app = FastAPI()
def common_params(q: str | None = None, skip: int = 0, limit: int = 10):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items")
def list_items(commons: dict = Depends(common_params)):
return commons
@app.get("/users")
def list_users(commons: dict = Depends(common_params)):
return commons类依赖
带 __init__ 的类可作为依赖项；FastAPI 像调用函数一样调用它。使用 Depends(CommonQueryParams) 或在参数标注为该类时使用简写 Depends()。实例会被注入。
from fastapi import FastAPI, Depends
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 10):
self.q = q
self.skip = skip
self.limit = limit
app = FastAPI()
@app.get("/items")
def list_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
# shorthand: commons = Depends() infers the type嵌套依赖
依赖项本身可以依赖其他依赖项，形成依赖图。FastAPI 解析整棵树，每个结果按请求缓存，并传播异常。这非常适合分层认证（token -> user -> permissions）。
from fastapi import FastAPI, Depends, Header, HTTPException
def get_token(authorization: str = Header()):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="invalid header")
return authorization.removeprefix("Bearer ")
def get_current_user(token: str = Depends(get_token)):
if token != "admin-token":
raise HTTPException(status_code=403, detail="forbidden")
return {"username": "admin"}
app = FastAPI()
@app.get("/me")
def me(user: dict = Depends(get_current_user)):
return userYield 依赖（清理）
依赖生成器（yield）非常适合需要清理的资源：数据库会话、文件句柄、事务。yield 后的代码作为终结器运行，即使路由抛出异常也会执行。每个依赖项只能有一个 yield。
from fastapi import FastAPI, Depends
def get_db():
db = open_db_session() # acquire resource
try:
yield db # injected into the route
finally:
db.close() # always runs after the request
app = FastAPI()
@app.get("/items")
def list_items(db = Depends(get_db)):
return db.query("SELECT * FROM items")全局依赖
向 FastAPI()（或 APIRouter）传入 dependencies=[...] 可在每条路由上强制执行它们，而无需注入其返回值。适用于全局认证、限流或请求日志。
from fastapi import FastAPI, Depends, Header, HTTPException
def verify_token(x_token: str = Header()):
if x_token != "supersecret":
raise HTTPException(status_code=400, detail="X-Token missing")
def verify_key(x_key: str = Header()):
if x_key != "realkey":
raise HTTPException(status_code=400, detail="X-Key missing")
# applied to every route in the app
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get("/items")
def list_items():
return [{"item": "a"}] # token & key already checked按路由依赖
在装饰器（或 APIRouter）上使用 dependencies=[Depends(...)] 为特定路由要求依赖项而不使用其返回值。这样可以在保持路由签名整洁的同时仍强制执行检查。
from fastapi import FastAPI, Depends, Header, HTTPException
def verify_token(x_token: str = Header()):
if x_token != "secret":
raise HTTPException(status_code=401, detail="bad token")
app = FastAPI()
@app.get("/public")
def public():
return {"open": True}
@app.get("/private", dependencies=[Depends(verify_token)])
def private():
return {"secret": "data"}安全（HTTPBasic/OAuth2/JWT）
HTTP 基本认证
HTTPBasic 将 Authorization: Basic 头解析为用户名/密码。使用 secrets.compare_digest（而非 ==）以防止时序攻击。始终返回带 WWW-Authenticate: Basic 的 401，以便浏览器提示输入凭据。
import secrets
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
security = HTTPBasic()
app = FastAPI()
def get_current_user(credentials: HTTPBasicCredentials = Depends(security)):
correct_user = secrets.compare_digest(credentials.username, "admin")
correct_pass = secrets.compare_digest(credentials.password, "secret")
if not (correct_user and correct_pass):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
@app.get("/profile")
def profile(user: str = Depends(get_current_user)):
return {"user": user}OAuth2 密码承载
OAuth2PasswordBearer 声明路由需要 Bearer token；依赖项从 Authorization 头提取它。tokenUrl 告诉文档 UI 在哪里发送凭据。token 本身在验证之前只是一个字符串。
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()
@app.get("/me")
def me(token: str = Depends(oauth2_scheme)):
return {"token": token}
# client POSTs username+password to /token, receives an access token,
# then sends it as: Authorization: Bearer <token>创建 JWT Token
使用 python-jose（或 PyJWT）签名 token。'sub'（主题）声明通常保存用户名或用户 ID，'exp' 设置过期时间。保持 SECRET_KEY 长、随机并从环境加载——切勿提交到代码库。
from datetime import datetime, timedelta, timezone
from jose import jwt
SECRET_KEY = "keep-this-secret"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_minutes: int = 30) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=expires_minutes)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
# token = create_access_token({"sub": "alice"})
# -> eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...解码 JWT Token
jwt.decode 自动验证签名和 'exp' 声明，失败时抛出 JWTError。将依赖项映射为带 WWW-Authenticate: Bearer 的 401，以便客户端知道需提供有效 token。
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "keep-this-secret"
ALGORITHM = "HS256"
def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str | None = payload.get("sub")
if username is None:
raise credentials_exception
return username
except JWTError:
raise credentials_exception
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)密码哈希
切勿存储明文密码。passlib 配合 bcrypt（或 argon2）处理加盐和哈希。verify() 将明文输入与存储的哈希比较。pip install passlib[bcrypt] 安装 bcrypt 后端。
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(plain: str) -> str:
return pwd_context.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
# usage when issuing tokens:
def authenticate_user(username: str, password: str):
user = fake_db.get(username)
if not user or not verify_password(password, user.hashed_password):
return False
return userOAuth2 登录流程
OAuth2PasswordRequestForm 从表单编码的 POST 读取用户名/密码（标准 OAuth2 密码流程）。返回 access_token 和 token_type: 'bearer'。此端点就是 OAuth2PasswordBearer 中 tokenUrl 指向的端点。
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
app = FastAPI()
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form.username, form.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
token = create_access_token({"sub": user.username})
return {"access_token": token, "token_type": "bearer"}
# OAuth2PasswordRequestForm parses username+password from form data
# and powers the 'Authorize' button in the Swagger UI.中间件
添加中间件
@app.middleware('http') 注册一个包裹每个请求的函数。它必须 await call_next(request) 获取响应，然后可在返回前检查或修改它。中间件对所有路由生效。
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{duration:.4f}"
return response自定义头部中间件
中间件是为每个响应添加头部（安全头、服务器标识、追踪 ID）的正确位置。在 await call_next 之后修改响应对象。
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_banner(request: Request, call_next):
response = await call_next(request)
response.headers["X-Powered-By"] = "FastAPI"
response.headers["X-Frame-Options"] = "DENY"
return response
# every response now carries these headersBaseHTTPMiddleware
对于可复用、可配置的中间件，继承 BaseHTTPMiddleware 并用 app.add_middleware(MyMiddleware, **options) 注册。内置的 CORS、GZip 和 TrustedHost 中间件也是这样添加的。
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
response.headers["X-Duration"] = str(time.perf_counter() - start)
return response
app = FastAPI()
app.add_middleware(TimingMiddleware)
# class-based middleware is reusable and configurableHTTPS 重定向
HTTPSRedirectMiddleware 将任何 HTTP 请求 308 重定向到 HTTPS 版本。在终止 TLS 的负载均衡器后面使用，以确保客户端升级到安全连接。
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)
# all HTTP requests are redirected to HTTPS (308)可信主机
TrustedHostMiddleware 拒绝 Host 头不在 允许列表中的请求，防止 Host 头注入攻击。支持通配符如 '*.example.com'。始终为本地开发包含 'localhost'。
from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["example.com", "*.example.com", "localhost"],
)
# requests with a disallowed Host header get a 400 response中间件顺序
Starlette 对入站请求以 LIFO（后进先出）顺序执行中间件：最后添加的最先运行。尽早添加 CORS（以便预检工作），GZip 在其后。调试某个中间件设置的头部为何在另一个中不可见时要记住这点。
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# Middleware runs in REVERSE order of registration for requests,
# and in registration order for responses.
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(CORSMiddleware, allow_origins=["*"])
# request flow: CORS -> GZip -> route
# response flow: route -> GZip -> CORSCORS
CORSMiddleware 设置
CORSMiddleware 处理 CORS 预检（OPTIONS）并添加 Access-Control-* 头。allow_origins 应列出允许的确切源；生产中使用具体 URL 列表而非 ['*']。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com", "https://www.example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/data")
def data():
return {"data": 1}允许所有源（仅开发）
allow_origins=['*'] 允许任何源，但不能与 allow_credentials=True 组合（浏览器会拒绝）。仅用于开发中的完全公共 API；生产中应枚举真实的源。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
# allow_credentials=True CANNOT be combined with allow_origins=["*"]
)允许凭据
allow_credentials=True 让浏览器跨源发送 cookie 和 Authorization 头。它需要具体的源（绝不能是 '*'）。浏览器在响应中设置 Access-Control-Allow-Credentials: true。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"],
allow_credentials=True, # send cookies / Authorization
allow_methods=["GET", "POST"],
allow_headers=["Authorization", "Content-Type"],
)
# needed when the frontend sends cookies or credentials with requests允许的方法与头
allow_methods/allow_headers 限制浏览器可发送的内容。expose_headers 列出前端 JavaScript 可读取的响应头（浏览器否则会隐藏自定义头）。预检响应由浏览器缓存。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"],
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
expose_headers=["X-Total-Count", "X-Process-Time"],
)
# expose_headers lets the frontend JS read these response headers预检请求
对于"非简单"请求（如自定义头、PUT/DELETE、JSON 内容类型），浏览器发送 OPTIONS 预检。CORSMiddleware 拦截 OPTIONS 并以允许的方法/头响应。max_age 控制浏览器缓存该答案的时间。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"],
allow_methods=["*"],
allow_headers=["*"],
max_age=600, # browser caches preflight result for 600s
)
# Browser sends OPTIONS preflight for non-simple requests;
# CORSMiddleware answers automatically with the right headers.动态源验证
对于动态允许列表，使用 allow_origin_regex（与请求 Origin 匹配）而非静态列表。当有效源遵循某种模式时（如 example.com 的任何子域）非常方便。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
VALID_DOMAINS = {"example.com", "staging.example.com"}
def origin_check(origin: str) -> str | None:
if origin and origin.replace("https://", "").replace("http://", "") in VALID_DOMAINS:
return origin
return None
app.add_middleware(
CORSMiddleware,
allow_origin_regex=r"https://.*\.example\.com",
allow_methods=["*"],
allow_headers=["*"],
)异常处理
抛出 HTTPException
在路由或依赖项中的任何位置 raise HTTPException 可短路并返回错误响应。detail 可以是字符串、dict 或 list——它成为 JSON 的 'detail' 字段。headers 可选地添加响应头。
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
items = {1: "Apple", 2: "Banana"}
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in items:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
headers={"X-Error": "missing"},
)
return {"item": items[item_id]}自定义异常
用 @app.exception_handler(MyException) 注册处理器，将任何自定义异常转换为 HTTP 响应。处理器接收请求和异常。这集中了应用领域错误的错误格式化。
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
@app.exception_handler(UnicornException)
async def unicorn_handler(request: Request, exc: UnicornException):
return JSONResponse(
status_code=418,
content={"message": f"Oops, {exc.name} did it again"},
)
@app.get("/unicorns/{name}")
def read_unicorn(name: str):
if name == "yolo":
raise UnicornException(name=name)
return {"name": name}覆盖验证错误
当输入未通过 Pydantic 验证时引发 RequestValidationError。用 @app.exception_handler(RequestValidationError) 覆盖其处理器以自定义 422 响应结构，例如匹配公司统一的错误格式。
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"success": False,
"errors": exc.errors(),
"body": exc.body,
},
)覆盖 HTTPException 处理器
覆盖默认的 HTTPException 处理器以在所有引发的 HTTPException 上强制一致的错误 JSON 结构。重新绑定时将类导入为 fastapi.exceptions.HTTPException（与 raise 使用的相同）。
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException as FastAPIHTTPException
app = FastAPI()
@app.exception_handler(FastAPIHTTPException)
async def custom_http_handler(request, exc: FastAPIHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"success": False,
"status": exc.status_code,
"message": exc.detail,
},
headers=exc.headers,
)常见状态码
status 模块为所有 HTTP 状态码提供命名常量，避免魔术数字。200（默认 GET）、201（创建）、204（无内容）、422（验证）是 FastAPI 应用中最常见的。
from fastapi import FastAPI, status
app = FastAPI()
@app.get("/ok", status_code=status.HTTP_200_OK)
def ok(): return {"ok": True}
@app.post("/create", status_code=status.HTTP_201_CREATED)
def create(): return {"id": 1}
@app.get("/no-content", status_code=status.HTTP_204_NO_CONTENT)
def no_content(): return None
@app.get("/not-found")
def missing():
# raise HTTPException(status.HTTP_404_NOT_FOUND, "nope")
...
# Other common: 400 BAD_REQUEST, 401 UNAUTHORIZED,
# 403 FORBIDDEN, 409 CONFLICT, 422 UNPROCESSABLE_ENTITY, 500 INTERNAL_SERVER_ERROR通用异常处理器
为基类 Exception 注册处理器可捕获任何尚未处理的异常，防止堆栈跟踪泄露给客户端。始终在服务端记录完整回溯；向客户端返回通用的 500。
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
logger = logging.getLogger("uvicorn.error")
app = FastAPI()
@app.exception_handler(Exception)
async def unhandled(request: Request, exc: Exception):
logger.exception("Unhandled error on %s", request.url.path)
return JSONResponse(
status_code=500,
content={"detail": "Internal Server Error"},
)
@app.get("/boom")
def boom():
raise RuntimeError("something exploded")后台任务
基本 BackgroundTasks
BackgroundTasks 在响应发送后运行函数——非常适合即发即忘的工作，如日志、邮件或缓存失效。add_task 调度函数（及参数）而不阻塞响应。
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message + "\n")
@app.post("/send")
def send_notification(background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, "notification sent")
return {"message": "notification queued"}多个任务
可添加任意数量的任务；它们在响应返回后按添加顺序依次运行。由于在同一进程内运行，应保持快速和非阻塞，或迁移到真正的队列（Celery、RQ、Dramatiq）。
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def send_email(to: str, subject: str): ...
def update_cache(key: str): ...
def notify_slack(msg: str): ...
@app.post("/signup")
def signup(background_tasks: BackgroundTasks):
background_tasks.add_task(send_email, "[email protected]", "Welcome")
background_tasks.add_task(update_cache, "user_count")
background_tasks.add_task(notify_slack, "new signup")
return {"status": "ok"}带依赖的任务
BackgroundTasks 像任何依赖项一样注入。任务函数可接收在请求时捕获的值（如数据库会话）。注意依赖项中 yield/关闭的资源在任务完成后关闭，因为任务在请求生命周期的范围内运行。
from fastapi import FastAPI, BackgroundTasks, Depends
app = FastAPI()
def get_db_session():
db = "session"
yield db
def write_log(db, message: str):
print(db, message)
@app.get("/items")
def list_items(background_tasks: BackgroundTasks, db = Depends(get_db_session)):
background_tasks.add_task(write_log, db, "list called")
return {"items": []}异步任务函数
后台任务函数可以是同步（在线程池中运行）或异步（在事件循环上 await）。对于自身使用异步库的 I/O 密集型工作使用异步任务，对于 CPU/阻塞型工作使用同步任务。
from fastapi import FastAPI, BackgroundTasks
import httpx
app = FastAPI()
async def fetch_url(url: str):
async with httpx.AsyncClient() as client:
r = await client.get(url)
return r.status_code
@app.post("/webhook")
def webhook(url: str, background_tasks: BackgroundTasks):
background_tasks.add_task(fetch_url, url)
return {"queued": url}Background 与 Celery 对比
对轻量、非关键、重启后丢失也无所谓的工作使用 BackgroundTasks。对于必须幸免于崩溃、可重试或跨机器扩展的任务，使用真正的任务队列（Celery、RQ、Dramatiq、Arq）配合 Redis 或 RabbitMQ 等 broker。
# BackgroundTasks: in-process, no broker, lost if process crashes
@app.post("/quick")
def quick(tasks: BackgroundTasks):
tasks.add_task(small_job)
return {"ok": True}
# Celery/RQ/Dramatiq: separate workers, persistent, retryable
# from celery import Celery
# @celery.task
# def big_job(payload): ...
#
# @app.post("/heavy")
# def heavy(payload: dict):
# big_job.delay(payload)
# return {"queued": True}发送邮件示例
发送邮件是 BackgroundTasks 的经典用例：它很慢，用户不应等待。响应立即返回， 邮件在之后发送。对于高发送量，建议使用队列和带重试的邮件服务 SDK。
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr
class Message(BaseModel):
to: EmailStr
subject: str
body: str
app = FastAPI()
def send_email(to: str, subject: str, body: str):
# smtplib or your provider's SDK
import smtplib
with smtplib.SMTP("localhost") as s:
s.sendmail("[email protected]", to, f"Subject: {subject}\n\n{body}")
@app.post("/contact")
def contact(msg: Message, tasks: BackgroundTasks):
tasks.add_task(send_email, msg.to, msg.subject, msg.body)
return {"status": "queued", "to": msg.to}WebSocket
WebSocket 基础
使用 @app.websocket() 定义 WebSocket 端点。在发送/接收前必须 accept() 连接。函数通常循环读取消息；receive_text/receive_bytes/receive_json 都 await 来自客户端的数据。
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
await ws.accept()
while True:
data = await ws.receive_text()
await ws.send_text(f"echo: {data}")
# client (browser):
# const ws = new WebSocket("ws://localhost:8000/ws");
# ws.onmessage = (e) => console.log(e.data);发送 JSON
send_json() 和 receive_json() 为你处理序列化，通过 socket 交换 JSON 对象。二进制数据使用 send_bytes/receive_bytes。每次发送/接收是 WebSocket 上的一帧。
import json
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await ws.accept()
while True:
msg = await ws.receive_json()
await ws.send_json({"reply": msg.get("text", "").upper()})
# send_text / send_bytes / send_json helpers are availableWebSocket 关闭
客户端关闭连接时引发 WebSocketDisconnect——捕获它以进行清理。调用 ws.close(code=1000) 进行正常关闭；常见关闭码有 1000（正常）、1008（策略违规）、1011（服务器错误）。
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await ws.accept()
try:
while True:
data = await ws.receive_text()
if data == "quit":
await ws.close(code=1000)
break
await ws.send_text(data)
except WebSocketDisconnect:
print("client disconnected")连接管理器
ConnectionManager 跟踪活动 socket，以便向所有连接的客户端广播（如聊天室）。务必在断开连接时移除 socket，避免向已关闭的连接发送数据并保持内存有界。
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active: list[WebSocket] = []
async def connect(self, ws: WebSocket):
await ws.accept()
self.active.append(ws)
def disconnect(self, ws: WebSocket):
self.active.remove(ws)
async def broadcast(self, message: str):
for ws in self.active:
await ws.send_text(message)
manager = ConnectionManager()
app = FastAPI()
@app.websocket("/chat")
async def chat(ws: WebSocket):
await manager.connect(ws)
try:
while True:
msg = await ws.receive_text()
await manager.broadcast(msg)
except WebSocketDisconnect:
manager.disconnect(ws)通过查询参数进行 WebSocket 认证
浏览器无法在 WebSocket 连接上设置自定义头，因此认证 token 通常作为查询参数（或 cookie）传递。在调用 accept() 之前验证，未授权时以代码 1008（策略违规）关闭。
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
app = FastAPI()
async def authenticate(token: str) -> str:
if token != "secret":
raise HTTPException(status_code=401)
return "alice"
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
token = ws.query_params.get("token")
if not token:
await ws.close(code=1008)
return
user = await authenticate(token)
await ws.accept()
try:
while True:
data = await ws.receive_text()
await ws.send_text(f"{user}: {data}")
except WebSocketDisconnect:
pass带依赖的 WebSocket
依赖项可在 WebSocket 端点中使用，就像在 HTTP 路由中一样，但只有从初始握手（头、查询参数、cookie）读取的依赖项才有意义——没有请求体。像平常一样通过 Depends() 注入。
from fastapi import FastAPI, WebSocket, Depends, Header
app = FastAPI()
def token_validator(token: str = Header(default="")):
# for WS, headers come from the initial HTTP handshake
return token == "valid"
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket, valid: bool = Depends(token_validator)):
if not valid:
await ws.close(code=1008)
return
await ws.accept()
while True:
msg = await ws.receive_text()
await ws.send_text(msg)数据库（SQLAlchemy）
SQLAlchemy 设置
create_engine 设置数据库连接池；SessionLocal 是会话工厂；Base 是模型的声明式基类。对于 FastAPI 中 的 SQLite，设置 check_same_thread=False，因为请求可能在不同线程中运行。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
# postgresql: "postgresql://user:pass@localhost:5432/dbname"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}, # sqlite only
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()会话依赖
yield 依赖是提供数据库会话的规范方式：每个请求获得一个在 finally 块中关闭的新会话。即使路由抛出异常也能保证清理。
from fastapi import Depends
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items")
def list_items(db: Session = Depends(get_db)):
return db.query(Item).all()
# each request gets its own session, closed automatically afterward模型定义
将表定义为继承 Base 的 SQLAlchemy ORM 模型。生产中使用 Alembic 进行 schema 迁移而非 create_all()。列可以建立索引、可空，并在数据库或 Python 级别设置默认值。
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
price = Column(Integer, default=0)
in_stock = Column(Boolean, default=True)
# create tables (for dev; use Alembic migrations in prod)
Base.metadata.create_all(bind=engine)创建记录
add() 暂存新行，commit() 写入数据库，refresh() 重新加载实例以获取服务器生成的值（如自增 id）。将多次写入包装在事务中并在最后提交一次。
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
@app.post("/items", status_code=201)
def create_item(payload: ItemCreate, db: Session = Depends(get_db)):
item = Item(name=payload.name, price=payload.price)
db.add(item)
db.commit()
db.refresh(item) # load generated id
return item查询记录
db.query(Model) 开始查询；.filter() 应用 WHERE 条件，.offset()/.limit() 分页，.first() 返回一行或 None，.all() 返回列表。简单相等条件使用 .filter_by(name=x)。
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
app = FastAPI()
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
return db.query(Item).offset(skip).limit(limit).all()
@app.get("/items/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if not item:
raise HTTPException(status_code=404, detail="not found")
return item更新与删除
对于部分更新，遍历 model_dump(exclude_unset=True) 使只 有提供的字段更改。delete() 暂存删除；commit() 最终确认。DELETE 成功时返回 204 No Content。
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
app = FastAPI()
@app.put("/items/{item_id}")
def update_item(item_id: int, payload: ItemUpdate, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if not item:
raise HTTPException(status_code=404, detail="not found")
for k, v in payload.model_dump(exclude_unset=True).items():
setattr(item, k, v)
db.commit()
db.refresh(item)
return item
@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if item:
db.delete(item)
db.commit()数据库（Tortoise ORM）
Tortoise 设置
Tortoise ORM 是异步原生的，与 FastAPI 配合良好。在启动时用 db_url 和包含模型的模块初始化它。generate_schemas() 适用于开发；生产中使用 Aerich 进行迁移。新代码中使用 lifespan 处理器代替 on_event。
from fastapi import FastAPI
from tortoise import Tortoise
app = FastAPI()
async def init():
await Tortoise.init(
db_url="sqlite://db.sqlite3",
modules={"models": ["app.models"]},
)
await Tortoise.generate_schemas()
@app.on_event("startup")
async def startup():
await init()
@app.on_event("shutdown")
async def shutdown():
await Tortoise.close_connections()模型定义
Tortoise 模型继承 Model 并通过 tortoise.fields 声明字段。字段类型映射到数据库列类型。在内部 Meta 类中设置表名。关系使用 ForeignKeyField、ManyToManyField 等。
from tortoise import fields
from tortoise.models import Model
class Item(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100)
price = fields.FloatField(default=0.0)
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "items"
def __str__(self):
return self.name创建记录
Item.create() 是一个插入并返回新行的协程。要返回 Pydantic 响应，用 from_tortoise_orm()（通过 tortoise.contrib.pydantic）构建 BaseModel，以便 Tortoise 正确序列化异步加载的关系。
from fastapi import FastAPI
from pydantic import BaseModel
from .models import Item
app = FastAPI()
class ItemIn(BaseModel):
name: str
price: float
@app.post("/items", response_model=ItemOut, status_code=201)
async def create_item(payload: ItemIn):
item = await Item.create(name=payload.name, price=payload.price)
return await ItemOut.from_tortoise_orm(item)异步查询
Tortoise 查询需要 await：.all() 返回查询集，.filter() 添加条件，.get_or_none() 获取一行或 None，.first() 返回第一个 匹配。除非确定只有一行，否则避免 .get()（找不到时会抛出异常）。
from fastapi import FastAPI, HTTPException
from .models import Item
app = FastAPI()
@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
return await Item.all().offset(skip).limit(limit)
@app.get("/items/{item_id}")
async def get_item(item_id: int):
item = await Item.get_or_none(id=item_id)
if not item:
raise HTTPException(status_code=404, detail="not found")
return item更新与删除
修改模型属性后 await save() 更新行，或 await Item.filter(id=...).update(name=x) 进行批量更新。await item.delete() 删除行。每个方法都是协程，必须 await。
from fastapi import FastAPI, HTTPException
from .models import Item
app = FastAPI()
@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str):
item = await Item.get_or_none(id=item_id)
if not item:
raise HTTPException(status_code=404, detail="not found")
item.name = name
await item.save()
return item
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
item = await Item.get_or_none(id=item_id)
if item:
await item.delete()Tortoise Pydantic Schema
tortoise.contrib.pydantic 直接从 Tortoise 模型生成 Pydantic 模型，因此无需维护两套 schema。单个对象使用 pydantic_model_creator，列表响应使用 pydantic_queryset_creator。
from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator
from .models import Item
ItemOut = pydantic_model_creator(Item, name="ItemOut")
ItemList = pydantic_queryset_creator(Item)
@app.get("/items", response_model=ItemList)
async def list_items():
return await Item.all()
# auto-generates a Pydantic model from the Tortoise model,
# including types and docstrings.测试（TestClient）
TestClient 基础
TestClient 包裹你的应用，让你在进程内发起请求（无真实 socket）。它基于 httpx，开箱即用地与 pytest 配合。导入与生产中运行的相同 app 实例。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}
# TestClient is based on httpx; no network is involved —
# it calls the ASGI app directly in-process.测试 GET 和 POST
json= 传 JSON 体，params= 传查询参数，headers= 传头部，files= 传 multipart 上传。验证失败返回 422——对 status_code 和错误 'detail' 结构进行断言。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_item():
response = client.post(
"/items",
json={"name": "Apple", "price": 0.5},
)
assert response.status_code == 201
assert response.json()["name"] == "Apple"
def test_validation_error():
response = client.post("/items", json={"name": "Apple"}) # missing price
assert response.status_code == 422Pytest Fixture
返回 TestClient 的 pytest fixture 让测试保持 DRY，每个测试获得全新的 client。对于数据库测试，添加创建/删除表或每测试事务回滚的 fixture 以保持隔离。
import pytest
from fastapi.testclient import TestClient
from main import app
@pytest.fixture
def client():
return TestClient(app)
def test_read(client):
r = client.get("/items")
assert r.status_code == 200
def test_create(client):
r = client.post("/items", json={"name": "X"})
assert r.status_code == 201
# run: pytest -v覆盖依赖
app.dependency_overrides 在测试中将依赖项（如真实数据库会话或认证）替换为 mock/fake，而无需更改路由。这是从外部服务隔离测试端点的规范方式。测试后清除它。
from fastapi.testclient import TestClient
from main import app, get_db
def get_test_db():
# return a test session or mock
yield FakeSession()
app.dependency_overrides[get_db] = get_test_db
client = TestClient(app)
def test_items():
r = client.get("/items")
assert r.status_code == 200
# restore afterwards:
app.dependency_overrides.clear()测试认证
OAuth2PasswordRequestForm 期望表单编码数据（data=，而非 json=）。在后续受保护请求的 Authorization: Bearer 头中使用返回的 token。要完全绕过认证，改为覆盖 get_current_user 依赖项。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def get_token():
r = client.post(
"/token",
data={"username": "admin", "password": "secret"},
)
return r.json()["access_token"]
def test_protected():
token = get_token()
r = client.get(
"/me",
headers={"Authorization": f"Bearer {token}"},
)
assert r.status_code == 200
assert r.json()["username"] == "admin"测试 WebSocket
TestClient.websocket_connect() 在上下文管理器内打开 WebSocket。使用 send_text/receive_text（及 json/bytes 变体）交换消息。在已关闭的连接上接收会引发 WebSocketDisconnect。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_websocket():
with client.websocket_connect("/ws") as ws:
ws.send_text("hello")
data = ws.receive_text()
assert data == "echo: hello"
def test_websocket_close():
with client.websocket_connect("/ws") as ws:
ws.send_text("quit")
# the server closes the connectionOpenAPI 文档
Swagger UI
FastAPI 根据你的路由、类型和文档字符串自动在 /docs 生成交互式 Swagger UI。在 app 上设置 title/description/version，并在每个路由添加文档字符串或 summary/description 以丰富 UI。
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="Sample API with auto docs",
version="1.0.0",
)
@app.get("/items", summary="List items", description="Returns all items")
def list_items():
"""Get every item in the catalog."""
return [{"id": 1}]
# interactive docs at: http://localhost:8000/docsReDoc
/redoc 的 ReDoc 是 Swagger 的只读、以文档为中心的替代方案。原始 OpenAPI schema 位于 /openapi.json，可输入到其他工具。用 docs_url=None、redoc_url=None、openapi_url=None 禁用文档（如生产环境）。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
return {"hello": "world"}
# ReDoc (alternative docs UI) at: http://localhost:8000/redoc
# raw OpenAPI JSON at: http://localhost:8000/openapi.json
# disable docs entirely:
# app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)标签与分组
标签在文档 UI 中对路由分组。在 openapi_tags 中预先声明标签元数据（name、description、externalDocs），使分组在任何路由使用它之前就出现。设置 deprecated=True 在文档中标记端点为已弃用。
from fastapi import FastAPI
app = FastAPI(openapi_tags=[
{"name": "users", "description": "User management"},
{"name": "items", "description": "Item operations"},
])
@app.get("/users", tags=["users"])
def list_users(): ...
@app.post("/items", tags=["items"])
def create_item(): ...
@app.get("/items/{id}", tags=["items"], deprecated=True)
def get_item(id: int): ...路由元数据
装饰器接受流入 OpenAPI 的元数据：summary（短标题）、description（长文本）、response_description、tags、deprecated 和 status_code。如未显式传入，文档字符串用作 description。
from fastapi import FastAPI, status
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
app = FastAPI()
@app.post(
"/items",
response_model=Item,
status_code=status.HTTP_201_CREATED,
summary="Create an item",
description="Creates a new item and returns it.",
tags=["items"],
response_description="The created item",
)
def create_item(item: Item):
"""Create endpoint docstring — also shown in docs."""
return item字段示例
Field(examples=[...]) 和 Body(examples=[...]) 为 Swagger UI 的"试用"和示例下拉添加示例载荷。Body 示例可包含多个带 summary 的用例，以展示有效和无效输入。
from pydantic import BaseModel, Field
from fastapi import FastAPI, Body
class Item(BaseModel):
name: str = Field(examples=["Apple"])
price: float = Field(examples=[0.5], gt=0)
app = FastAPI()
@app.post("/items")
def create_item(item: Item):
return item
@app.post("/manual")
def manual(
item: Item = Body(
examples=[
{"summary": "valid", "value": {"name": "Apple", "price": 0.5}},
{"summary": "invalid", "value": {"name": "Apple", "price": -1}},
],
),
):
return item自定义 OpenAPI Schema
覆盖 app.openapi 以自定义生成的 schema——添加 logo、额外元数据或移除/转换操作。将结果缓存在 app.openapi_schema 中使其只生成一次。该 schema 就是 /openapi.json 提供的内容。
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
app = FastAPI()
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
schema = get_openapi(
title="Custom API",
version="2.0.0",
routes=app.routes,
)
schema["info"]["x-logo"] = {"url": "https://example.com/logo.png"}
app.openapi_schema = schema
return schema
app.openapi = custom_openapi部署（Uvicorn/Gunicorn）
Uvicorn 生产环境
生产环境中，在反向代理（Nginx、Caddy、云负载均衡器）后面运行 Uvicorn。使用 --proxy-headers 让应用从 X-Forwarded-For/X-Forwarded-Proto 看到真实客户端 IP/主机。生产中避免 --reload。
# Run directly with uvicorn (single worker)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1
# Recommended: behind a reverse proxy (nginx) on port 8000
# Use --proxy-headers so X-Forwarded-* is trusted
uvicorn main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips='*'Gunicorn 配合 Uvicorn Worker
Gunicorn 管理多个 Uvicorn worker 进程，提供优雅重载、崩溃重启和预 fork 扩展。使用 UvicornWorker 类。worker 数量以 (2 * CPU 核心数 + 1) 为起点，然后基准测试。
# install the uvicorn worker class
pip install "uvicorn[standard]" gunicorn
# run 4 worker processes managed by gunicorn
gunicorn main:app \
-w 4 \
-k uvicorn.workers.UvicornWorker \
-b 0.0.0.0:8000 \
--timeout 120
# -w workers (commonly 2*CPU+1), -k worker classDocker 部署
精简的 Python 基础镜像使镜像更小。在复制代码之前安装依赖以利用 Docker 的层缓存。在容器内用 gunicorn+uvicorn worker 运行，并将 8000 端口暴露给宿主机或编排器。
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "main:app", "-w", "4", \
"-k", "uvicorn.workers.UvicornWorker", \
"-b", "0.0.0.0:8000"]
# build & run:
# docker build -t myapi .
# docker run -p 8000:8000 myapi环境配置
pydantic-settings（原 pydantic.BaseSettings）从环境变量和 .env 文件加载配置并进行类型验证。这使同一镜像仅通过更改环境变量即可在不同环境（开发/预发布/生产）中运行。
import os
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str = "sqlite:///./app.db"
secret_key: str = "change-me"
debug: bool = False
allowed_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
settings = Settings()
# .env file:
# DATABASE_URL=postgresql://user:pass@db:5432/app
# SECRET_KEY=long-random-string
# DEBUG=falseNginx 反向代理
Nginx 终止 HTTP(S) 并转发到 localhost:8000 的 Uvicorn。设置 X-Forwarded-* 头（并在 Uvicorn 上使用 --proxy-headers）使应用看到真实客户端 IP 和 scheme。Upgrade/Connection 行启用 WebSocket 支持。
# /etc/nginx/conf.d/api.conf
server {
listen 80;
server_name api.example.com;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host \$host;
proxy_set_header X-Real-IP \$remote_addr;
proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto \$scheme;
proxy_http_version 1.1;
proxy_set_header Upgrade \$http_upgrade;
proxy_set_header Connection "upgrade";
}
}扩展与并发
两种扩展方式：每台机器更多 Uvicorn worker 进程（CPU 密集型），以及负载均衡器后面的更多机器（水平扩展）。对于异步应用，事件循环每个 worker 处理许多并发 I/O 请求。添加 /health 端点以便编排器（k8s、ECS）探测存活。
# Workers = process count (vertical scaling); pick ~2*CPU
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker
# Inside async app: concurrency is cooperative via the event loop.
# CPU-bound work blocks it — offload to a threadpool or task queue.
# Horizontal scaling: run several containers behind a load balancer.
# Use a shared store (Redis) for session state and rate limits.
# Health check endpoint for the balancer:
@app.get("/health")
def health():
return {"status": "ok"}进阶主题
Lifespan 事件
lifespan 上下文管理器替代已弃用的 on_event('startup'/'shutdown') 处理器。yield 之前的代码在启动时运行；yield 之后的代码在关闭时运行，即使启动失败也会执行。用它初始化数据库连接、ML 模型和缓存。
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# runs at startup
print("starting up")
app.state.cache = {}
yield
# runs at shutdown
print("shutting down")
await app.state.cache.clear()
app = FastAPI(lifespan=lifespan)
# the old @app.on_event("startup") / "shutdown" are deprecated
# in favor of the lifespan handler above.子应用（Mount）
app.mount() 在路径前缀下挂载子应用，适用于版本化 API 或组合多个应用。每个子应用有自己的路由和 OpenAPI schema；父应用的文档默认不包含子应用的路由。
from fastapi import FastAPI
app = FastAPI()
api_v1 = FastAPI()
api_v2 = FastAPI()
@api_v1.get("/items")
def v1_items():
return {"version": 1}
@api_v2.get("/items")
def v2_items():
return {"version": 2}
app.mount("/v1", api_v1)
app.mount("/v2", api_v2)
# GET /v1/items -> {"version": 1}
# GET /v2/items -> {"version": 2}APIRouter
APIRouter 将相关路由分组到一个模块中，共享前缀、标签、依赖项和响应。通过 app.include_router(router) 引入（可自带前缀或标签）。这是构建非平凡应用的标准模式。
from fastapi import APIRouter, Depends
router = APIRouter(
prefix="/items",
tags=["items"],
dependencies=[Depends(verify_token)],
responses={404: {"description": "Not found"}},
)
@router.get("/")
def list_items():
return []
@router.get("/{item_id}")
def get_item(item_id: int):
return {"id": item_id}
# in main.py: app.include_router(router)静态文件
StaticFiles 在 URL 前缀下提供静态资源目录（图片、CSS、JS）。每个目录挂载一次。对于 SPA 前端，挂载一个指向 index.html 的最终全匹配路由。静态文件为速度绕过路由系统。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
app = FastAPI()
# serve files from ./static at /static
app.mount("/static", StaticFiles(directory="static"), name="static")
# /static/logo.png -> ./static/logo.png
# /static/css/app.css -> ./static/css/app.css流式响应
StreamingResponse 发送按块生成的响应（生成器/迭代器），而非在内存中构建整个 body——非常适合大型 CSV 导出、文件流或 SSE。适当设置 media_type 以便客户端正确处理流。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import csv, io
app = FastAPI()
def iter_csv():
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["id", "name"])
for i in range(1000):
writer.writerow([i, f"name-{i}"])
yield output.getvalue()
output.seek(0)
output.truncate(0)
@app.get("/export.csv")
def export():
return StreamingResponse(
iter_csv(),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=export.csv"},
)自定义路由类
继承 APIRoute 可用自定义逻辑（日志、追踪、计时）包裹每个路由处理器，而无需触碰每个端点。设置 app.router.route_class 或向 APIRouter 传 route_class。处理器对每个匹配的请求运行。
from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute
from typing import Callable
class LoggingRoute(APIRoute):
def get_route_handler(self) -> Callable:
original = super().get_route_handler()
async def handler(request: Request) -> Response:
print(f"-> {request.method} {request.url.path}")
response = await original(request)
print(f"<- {response.status_code}")
return response
return handler
app = FastAPI()
app.router.route_class = LoggingRoute
@app.get("/items")
def items():
return {"items": []}相关 FastAPI 代码片段
Copy-paste ready code for common tasks.
Path Parameters
Capture typed path segments with validation.
Query Parameters
Parse query strings with defaults and validation.
Pydantic Request Body
Validate JSON bodies with Pydantic models.
Dependency Injection
Share logic via Depends and yield-based dependencies.
Response Model
Shape and filter responses with response_model.
JWT Auth
Issue and verify JSON Web Tokens with OAuth2.
Async and Await
Define async handlers and run blocking work in a thread.
Middleware
Add CORS, timing, and custom middleware.
这篇内容对您有帮助吗？