Article

后端框架 FastAPI

更新于:2026-07-13

第一章:FastAPI 入门与环境搭建

1.1 FastAPI 简介与核心优势(异步、自动文档、类型提示)

方法/特性名称语法/说明用途代码示例注意事项
异步支持(async/await)async def func(): / await some_async_op()实现非阻塞 I/O,提升高并发性能async def read_data(): / data = await db.fetch() / return data适用于数据库、网络请求等耗时操作
自动生成 API 文档访问 /docs/redoc基于 OpenAPI 和 JSON Schema 生成交互式文档启动应用后访问:http://localhost:8000/docs无需额外配置,实时更新
类型提示(Type Hints)param: str, return -> int提高代码可读性,支持自动验证与补全def greet(name: str) -> str: / return f"Hello {name}"配合 Pydantic 实现请求/响应数据校验
Pydantic 集成继承 BaseModel定义请求体、响应体结构,自动验证数据class User(BaseModel): / name: str / age: int字段类型错误会自动返回 422 响应
高性能基于 Starlette 与 Pydantic框架本身开销小,支持高吞吐量性能接近 Node.js 和 Go适合构建微服务和高性能 API

1.2 安装 FastAPI 与运行服务器(Uvicorn)

方法/命令名称语法用途代码示例注意事项
安装 FastAPIpip install fastapi安装 FastAPI 核心框架pip install fastapi通常需搭配 ASGI 服务器使用
安装 Uvicornpip install "uvicorn[standard]"安装 ASGI 服务器用于运行应用pip install uvicorn[standard] 包含推荐的依赖(如 uvloop)
运行应用uvicorn app:app启动 Uvicorn 服务器uvicorn main:app --reloadapp:app 表示模块名:应用实例名
查看帮助uvicorn --help查看所有运行参数选项uvicorn --help可查看 host、port、workers 等配置
安装生产依赖pip install "fastapi[all]"一次性安装所有可选依赖pip install "fastapi[all]"包括 pydantic、uvicorn、jinja2 等

1.3 第一个 FastAPI 应用:Hello World

方法/组件名称语法用途代码示例注意事项
导入 FastAPIfrom fastapi import FastAPI引入框架核心类from fastapi import FastAPI必须第一步导入
创建应用实例app = FastAPI()初始化 FastAPI 应用app = FastAPI()通常命名为 app
定义 GET 路由@app.get("/")绑定函数到指定路径的 GET 请求@app.get("/") / def read_root(): / return {"Hello": "World"}装饰器语法,路径 / 表示根路径
返回字典return {"key": "value"}自动序列化为 JSON 响应return {"message": "OK"}支持 dict、list、str、int 等基本类型
启动服务器uvicorn.run()在代码中直接启动(可选)if __name__ == "__main__": / uvicorn.run(app, host="127.0.0.1", port=8000)多用于开发或调试,生产环境推荐命令行启动

1.4 启动配置:host、port、reload、debug 模式

配置项/参数语法用途代码示例注意事项
host--host 0.0.0.0指定服务器监听的主机地址uvicorn main:app --host 0.0.0.00.0.0.0 表示允许外部访问
port--port 8080指定服务器监听端口uvicorn main:app --port 8080默认为 8000
reload--reload开启热重载,代码修改自动重启uvicorn main:app --reload仅用于开发环境,生产禁用
debug 模式app = FastAPI(debug=True)启用调试模式,输出详细错误app = FastAPI(debug=True)--reload 配合使用更佳
workers--workers 4指定工作进程数(生产)uvicorn main:app --workers 4多进程提升并发处理能力,需配合 Gunicorn

1.5 项目结构与模块化初识

结构/方法名称语法/布局用途示例结构注意事项
基础项目结构. / ├── main.py / ├── routers/ / │ └── items.py / └── models/ / └── item.py组织代码,便于维护推荐使用包结构管理大型项目使用 __init__.py 可选(Python 3.3+)
APIRouterfrom fastapi import APIRouter / router = APIRouter() / @router.get("/items") / def read_items(): ...将路由分组,实现模块化from fastapi import APIRoutermain.py 中通过 app.include_router() 注册
include_routerapp.include_router(router)将子路由挂载到主应用app.include_router(router, prefix="/api/v1", tags=["items"])可设置统一前缀、标签等元信息
模块导入from .routers import items相对导入子模块from .routers.items import router as items_router / app.include_router(items_router)确保 PYTHONPATH 正确或使用 -m 运行
主入口文件main.py应用启动入口包含 FastAPI() 实例和路由注册通常不包含具体业务逻辑

第二章:路由与路径操作函数

2.1 使用 @app.get()、@app.post() 等定义路由

方法名称语法用途代码示例注意事项
@app.get()@app.get("/path")定义处理 GET 请求的路由@app.get("/") / def read_root(): / return {"message": "Hello"}用于获取资源,不包含请求体
@app.post()@app.post("/path")定义处理 POST 请求的路由@app.post("/items") / def create_item(name: str): / return {"name": name}通常用于创建资源,支持请求体
@app.put()@app.put("/path")定义处理 PUT 请求的路由@app.put("/items/{item_id}") / def update_item(item_id: int, name: str): / return {"item_id": item_id, "name": name}用于完整更新资源
@app.delete()@app.delete("/path")定义处理 DELETE 请求的路由@app.delete("/items/{item_id}") / def delete_item(item_id: int): / return {"deleted": item_id}用于删除指定资源
@app.patch()@app.patch("/path")定义处理 PATCH 请求的路由@app.patch("/items/{item_id}") / def partial_update(item_id: int, name: str = None): / return {"item_id": item_id, "name": name}用于部分更新资源,非幂等

2.2 路径参数(Path Parameters)与类型声明

方法名称语法用途代码示例注意事项
路径参数定义{param_name}在 URL 中声明动态参数@app.get("/users/{user_id}")参数名需与函数参数一致
类型提示param_name: int / str / float自动转换路径参数类型def read_user(user_id: int): / return {"user_id": user_id}支持 int, str, float, bool 等
枚举类型param: Literal["value1", "value2"]限制路径参数取值范围from typing import Literal / @app.get("/cats/{name}") / def read_cat(name: Literal["garfield", "felix"]): / return {"name": name}使用 Literal 实现枚举约束
路径包含斜杠{file_path:path}捕获包含斜杠的完整路径@app.get("/files/{file_path:path}") / def read_file(file_path: str): / return {"file_path": file_path}:path 表示匹配任意长度路径

2.3 查询参数(Query Parameters)与可选值处理

方法名称语法用途代码示例注意事项
查询参数函数参数未在路径中从 URL 查询字符串提取参数@app.get("/items") / def list_items(skip: int = 0, limit: int = 10): / return {"skip": skip, "limit": limit}默认值决定是否为可选参数
可选参数param: Optional[str] = None显式声明可选查询参数from typing import Optional / def search(q: Optional[str] = None): / return {"q": q}必须导入 Optional
必填查询参数param: str = ...声明无默认值的必填查询参数def read_item(item_id: str, needy: str): / return {"item_id": item_id, "needy": needy}缺少时返回 422 错误
多值查询参数param: List[str] = Query([])接收多个同名查询参数from fastapi import Query / def tags(tags: list[str] = Query([])): / return {"tags": tags}需导入 Query 并设置默认值

2.4 多种 HTTP 方法支持(PUT、DELETE、PATCH 等)

方法名称语法用途代码示例注意事项
PUT@app.put("/item/{id}")替换整个资源@app.put("/items/{item_id}") / def update_item(item_id: int, item: Item): / return {"item_id": item_id, **item.dict()}幂等操作,完全更新
DELETE@app.delete("/item/{id}")删除资源@app.delete("/items/{item_id}") / def remove_item(item_id: int): / return {"status": "deleted"}通常无请求体
PATCH@app.patch("/item/{id}")部分更新资源@app.patch("/items/{item_id}") / def update_name(item_id: int, name: str): / return {"item_id": item_id, "name": name}非幂等,只更新提供字段
HEAD@app.head("/path")获取响应头信息(无正文)@app.head("/items") / def head_items(): / return {}自动由 GET 路由生成
OPTIONS@app.options("/path")获取资源支持的 HTTP 方法@app.options("/items") / def options_items(): / return {}框架自动处理,通常无需手动定义

2.5 路径顺序与优先级

概念说明用途代码示例注意事项
字面量路径优先先定义具体路径避免被通配路径拦截@app.get("/users/me") / def read_me(): ... / @app.get("/users/{user_id}") / def read_user(user_id: int): .../users/me 必须在 /users/{user_id} 之前
类型转换不影响优先级路径字符串匹配优先确保正确路由分发@app.get("/static/{path}") / def static_file(path: str): ... / @app.get("/static/data") / def data(): .../static/data 不会匹配到 path 参数
通配路径最后定义使用 {path:path} 匹配剩余路径实现静态文件或代理路由@app.get("/{full_path:path}") / def catch_all(full_path: str): / return {"path": full_path}必须放在所有路由最后

第三章:请求数据处理

3.1 使用 Pydantic 定义请求体模型(BaseModel)

方法名称语法用途代码示例注意事项
BaseModel 继承class Model(BaseModel):定义结构化数据模型class Item(BaseModel): / name: str / price: float所有字段需有类型注解
必填字段field: Type无默认值的字段为必填class User(BaseModel): / email: str / password: str请求中必须提供
可选字段field: Type = default设置默认值使字段可选class User(BaseModel): / is_active: bool = True可省略,使用默认值
高级类型List, Dict, Optional, Union使用复杂数据类型from typing import List, Optional / class Order(BaseModel): / items: List[str] / note: Optional[str] = None需从 typing 导入

3.2 处理 JSON 请求体(Body 参数)

方法名称语法用途代码示例注意事项
请求体参数def func(item: Item)接收 JSON 格式的请求体@app.post("/items") / def create_item(item: Item): / return item自动解析 JSON 并验证
多个请求体def func(a: ModelA, b: ModelB)接收多个模型(不推荐)FastAPI 不直接支持多个 Body需封装到一个模型中
嵌套请求体class Outer(BaseModel): / inner: InnerModel处理复杂嵌套结构class Item(BaseModel): / name: str / tags: list[str] = [] / @app.post("/items") / def create_item(item: Item): ...支持任意层级嵌套
Body 函数Body(..., embed=True)强制将单个参数包装为 JSON 对象@app.post("/items") / def create_item(name: str = Body(..., embed=True)): / return {"name": name}用于单个参数也需 JSON 包装的场景

3.3 获取查询参数(Query 参数)

方法名称语法用途代码示例注意事项
Query 函数param: str = Query(default)增强查询参数功能from fastapi import Query / def search(q: str = Query(None, min_length=3)): ...可设置验证、别名等
参数别名Query(..., alias="item-query")使用 URL 不友好参数名def read_item(item_query: str = Query(..., alias="item-query")): ...代码中用 Python 风格命名
多值参数param: List[str] = Query([])接收重复的查询参数def tags(tags: list[str] = Query([])): ...URL: ?tags=1&tags=2
参数验证Query(..., min_length=3)对查询参数进行约束def search(q: str = Query(..., min_length=3, max_length=50)): ...错误时返回 422

3.4 获取路径参数(Path 参数)

方法名称语法用途代码示例注意事项
Path 函数param: int = Path(...)为路径参数添加验证from fastapi import Path / def get_item(item_id: int = Path(..., ge=1)): ...... 表示必填
数值约束ge, gt, le, lt设置数值范围item_id: int = Path(..., ge=1, le=100)ge=大于等于,gt=大于
字符串约束min_length, max_length设置字符串长度name: str = Path(..., min_length=2)适用于路径中的字符串参数
参数描述Path(..., title="ID")添加参数元信息user_id: int = Path(..., title="用户ID")出现在文档中

3.5 获取 Cookie、Header 参数(Cookie、Header)

方法名称语法用途代码示例注意事项
Header 函数param = Header(...)从请求头中提取值from fastapi import Header / def get_token(x_token: str = Header(...)): ...自动处理 header 名称转换(如 x-token)
Cookie 函数param = Cookie(...)从 Cookie 中提取值from fastapi import Cookie / def get_session(session_id: str = Cookie(None)): ...session_id 可能为空
多个 Headerheaders: dict = Header()获取所有请求头def read_headers(headers: dict = Header(...)): ...返回字典形式的所有头信息
自定义 Header 名Header(..., convert_underscores=False)保留下划线(如 X_API_KEY)x_api_key: str = Header(..., convert_underscores=False)默认会将 _ 转为 -

3.6 表单数据处理(Form)

方法名称语法用途代码示例注意事项
Form 函数param = Form(...)从表单中提取字段from fastapi import Form / @app.post("/login") / def login(username: str = Form(...), password: str = Form(...)): ...需安装 python-multipart
必填表单字段Form(...)声明必填表单字段username: str = Form(...)缺少时返回 422
可选表单字段Form(default)设置默认值note: str = Form("")可省略
文件与表单混合file: UploadFile, field: str = Form(...)同时处理文件和表单字段@app.post("/upload") / def upload(file: UploadFile, desc: str = Form("")): ...Content-Type 为 multipart/form-data

3.7 文件上传(File 与 UploadFile)

方法名称语法用途代码示例注意事项
UploadFile 类param: UploadFile接收上传的文件对象from fastapi import UploadFile / def upload(file: UploadFile): / return {"filename": file.filename}支持异步读取
读取文件内容await file.read()异步读取文件二进制数据content = await file.read()大文件需分块读取
文件元信息file.filename, file.content_type获取文件名和 MIME 类型name = file.filename / mime = file.content_type可用于验证或存储
多文件上传files: List[UploadFile] = File(...)接收多个文件from typing import List / from fastapi import File / def upload_files(files: List[UploadFile] = File(...)): ...HTML 中 input 设置 multiple
File 函数File(..., description="...")增强文件参数功能image: UploadFile = File(..., description="上传头像")可设置验证、描述等

第四章:响应处理与模型定义

4.1 返回 JSON 响应(return dict 或 return BaseModel)

方法名称语法用途代码示例注意事项
返回字典return {"key": "value"}自动序列化为 JSON@app.get("/") / def home(): / return {"status": "ok"}支持嵌套字典和列表
返回 BaseModelreturn Model(**data)自动转为 JSON 并校验class Resp(BaseModel): / msg: str / @app.get("/", response_model=Resp) / def home(): / return Resp(msg="ok")推荐用于响应结构化
返回列表return [item1, item2]返回 JSON 数组return [{"id": 1}, {"id": 2}]适用于集合资源

4.2 使用 Response 类自定义响应

方法名称语法用途代码示例注意事项
Response 基类return Response(...)自定义响应内容与类型from starlette.responses import Response / return Response(content="data", media_type="text/csv")最基础的响应类
JSONResponsereturn JSONResponse(...)显式返回 JSON 响应from fastapi.responses import JSONResponse / return JSONResponse(content={"error": "not found"}, status_code=404)可自定义状态码和头
PlainTextResponsereturn PlainTextResponse(...)返回纯文本from fastapi.responses import PlainTextResponse / return PlainTextResponse("Hello", status_code=200)media_type=text/plain
HTMLResponsereturn HTMLResponse(...)返回 HTML 内容from fastapi.responses import HTMLResponse / return HTMLResponse("<h1>OK</h1>")media_type=text/html

4.3 设置响应状态码(status_code 参数)

方法名称语法用途代码示例注意事项
status_code 参数@app.post(..., status_code=201)自定义成功状态码@app.post("/items", status_code=201) / def create_item(item: Item): ...POST 创建常用 201
在函数中设置return JSONResponse(..., status_code=202)动态设置状态码return JSONResponse(content={"msg": "accepted"}, status_code=202)覆盖装饰器设置
重定向状态码status_code=302实现重定向return RedirectResponse(url="/new", status_code=302)配合 Location 头使用

4.4 响应模型(response_model)与字段过滤

方法名称语法用途代码示例注意事项
response_model@app.post(..., response_model=Model)指定响应数据结构@app.post("/users", response_model=UserOut) / def create_user(user: UserIn): ...自动过滤私密字段
exclude_unsetresponse_model_exclude_unset=True排除未设置的字段return JSONResponse(content=user.dict(exclude_unset=True), status_code=200)适用于 PATCH 响应
exclude_defaultsresponse_model_exclude_defaults=True排除默认值字段user.dict(exclude_defaults=True)减少响应体积
exclude_noneresponse_model_exclude_none=True排除值为 None 的字段user.dict(exclude_none=True)避免返回空值

4.5 返回纯文本、HTML、文件等响应类型

方法名称语法用途代码示例注意事项
PlainTextResponseresponse_class=PlainTextResponse返回纯文本@app.get("/", response_class=PlainTextResponse) / def home(): / return "Hello World"简化文本响应
HTMLResponseresponse_class=HTMLResponse返回 HTML 页面@app.get("/", response_class=HTMLResponse) / def home(): / return "<h1>Hi</h1>"可结合模板引擎
FileResponsereturn FileResponse("path.txt")返回静态文件from fastapi.responses import FileResponse / return FileResponse("data.csv")自动设置 Content-Type
StreamingResponsereturn StreamingResponse(...)流式传输大文件适用于视频、大 CSV 等避免内存溢出

4.6 处理响应头与 Cookie(Response 对象方法)

方法名称语法用途代码示例注意事项
set_header()response.headers["X-Auth"] = "val"设置响应头from fastapi import Response / def home(response: Response): / response.headers["X-Version"] = "1.0"response 是函数参数
set_cookie()response.set_cookie(...)设置 Cookieresponse.set_cookie(key="session_id", value="123")可设置过期时间、安全标志
delete_cookie()response.delete_cookie(...)删除 Cookieresponse.delete_cookie("session_id")实际是设置过期
CORS 头from fastapi.middleware.cors import CORSMiddleware处理跨域app.add_middleware(CORSMiddleware, allow_origins=["*"])需中间件支持

第五章:数据验证与错误处理

5.1 Pydantic 模型字段验证(Field 参数)

方法名称语法用途代码示例注意事项
Field 函数Field(...)添加字段级验证和元数据from pydantic import Field / name: str = Field(..., min_length=2, max_length=50)... 表示必填
字符串验证min_length, max_length, regex约束字符串格式email: str = Field(..., regex=".+@.+..com")可组合使用
数值验证gt, ge, lt, le约束数值范围age: int = Field(..., ge=0, le=150)ge=≥, gt=>, le=≤, lt=<
默认工厂default_factory=list可变默认值tags: List[str] = Field(default_factory=list)避免可变对象共享

5.2 查询/路径参数的约束(最小值、最大值、正则等)

方法名称语法用途代码示例注意事项
Query 验证Query(..., min_length=3)验证查询参数q: str = Query(..., min_length=3)错误返回 422
Path 验证Path(..., ge=1)验证路径参数item_id: int = Path(..., ge=1)路径参数也需验证
正则表达式Query(..., regex="^abc")模式匹配code: str = Query(..., regex="^[A-Z]{3}$")使用标准 Python 正则
列表长度Query(..., max_items=5)限制列表项数tags: List[str] = Query(..., max_items=5)适用于多值查询参数

5.3 自定义验证函数(@validator、@root_validator)

方法名称语法用途代码示例注意事项
@validator@validator('field')字段级自定义验证from pydantic import validator / @validator('price') / def price_must_be_positive(cls, v): / if v <= 0: / raise ValueError('must be > 0') / return vcls 是类,v 是值
@root_validator@root_validator(pre=True)跨字段验证@root_validator / def check_passwords_match(cls, values): / pw = values.get('password') / confirm = values.get('confirm') / if pw != confirm: / raise ValueError('passwords do not match') / return valuespre=True 在字段验证前执行
验证顺序pre=True / pre=False控制验证执行时机@root_validator(pre=False)pre=False 是默认,后执行

5.4 使用 HTTPException 抛出错误

方法名称语法用途代码示例注意事项
HTTPExceptionraise HTTPException(status_code, detail)抛出标准 HTTP 错误from fastapi import HTTPException / raise HTTPException(status_code=404, detail="Item not found")自动返回 JSON 错误响应
自定义状态码400, 401, 403, 404, 500根据错误类型选择raise HTTPException(403, "Forbidden")遵循 HTTP 语义
错误详情detail="描述信息"提供错误原因detail="User not authorized"可为字符串或字典

5.5 全局异常处理器(@app.exception_handler)

方法名称语法用途代码示例注意事项
@app.exception_handler()@app.exception_handler(ExceptionType)捕获特定异常@app.exception_handler(RequestValidationError) / def handle_validation_error(request, exc): / return JSONResponse(..., status_code=422)优先级高于默认处理器
处理自定义异常class CustomError(Exception): ...定义业务异常class InactiveUserError(Exception): pass / @app.exception_handler(InactiveUserError) / def handle_inactive(request, exc): ...统一处理业务错误
通用异常处理@app.exception_handler(Exception)捕获所有未处理异常用于记录日志并返回 500生产环境避免暴露敏感信息

5.6 验证错误的默认响应格式

特性格式/说明用途示例注意事项
默认错误状态码422 Unprocessable Entity请求数据验证失败status_code=422用于 Pydantic 验证错误
错误响应结构JSON 包含 loc, msg, type提供详细错误信息{"detail": [{"loc": ["body", "age"], "msg": "ensure this value is >= 0", "type": "value_error.number.not_ge"}]}loc 表示错误位置
错误位置 loc路径如 ["body", "user", "age"]指明错误字段层级便于前端定位问题字段支持 body, query, path, header
国际化支持可自定义错误消息适配多语言在 Field 或 validator 中设置需结合 i18n 库实现

第六章:依赖注入系统

6.1 依赖注入概念与优势

概念说明用途代码示例注意事项
依赖注入(DI)将功能逻辑解耦,由框架自动注入所需组件提高代码复用性、可测试性和可维护性def get_db(): / return db_session / @app.get("/items", dependencies=[Depends(get_db)])避免硬编码和全局变量
声明式编程通过声明依赖而非手动调用简化函数逻辑,专注业务处理使用 Depends() 显式声明依赖框架负责解析和执行
自动解析FastAPI 自动解析依赖层级支持嵌套依赖和参数传递支持函数、类、子依赖无需手动管理调用顺序

6.2 定义和使用函数依赖

方法名称语法用途代码示例注意事项
函数依赖def dependency(): ...定义可复用的验证或数据获取逻辑def require_auth(token: str = Header(...)): / if not valid(token): / raise HTTPException(401) / return token返回值可用于主函数参数
参数注入Depends(dependency_func)在路由中使用函数依赖@app.get("/secure", dependencies=[Depends(require_auth)]) / def secure_data(): ...可用于全局或单个路由
返回值传递return value将结果传递给主函数def get_user(token): / return db.get_user(token) / def route(user = Depends(get_user)): ...主函数可通过参数接收

6.3 类依赖与 call 方法

方法名称语法用途代码示例注意事项
类依赖class Dependency: __call__ = ...实现状态化或配置化依赖class RateLimiter: / def __init__(self, limit): / self.limit = limit / def __call__(self, ip: str = Header(...)): / if too_many_requests(ip, self.limit): / raise HTTPException(429)类需实现 __call__
初始化参数limiter = RateLimiter(100)创建带配置的依赖实例ten_per_min = RateLimiter(10) / @app.get("/data", dependencies=[Depends(ten_per_min)])支持复用不同配置
状态管理self.counter, self.config维护内部状态或配置可用于缓存、计数器等场景注意线程/异步安全

6.4 共享依赖(全局、路由、参数级)

方法名称语法用途代码示例注意事项
全局依赖FastAPI(dependencies=[...])为所有路由添加统一依赖app = FastAPI(dependencies=[Depends(check_api_key)])适用于认证、日志等通用逻辑
路由级依赖@app.get(..., dependencies=[...])为特定路由组添加依赖@app.get("/admin", dependencies=[Depends(is_admin)])比全局依赖优先级高
参数级依赖param = Depends(...)在函数参数中直接使用def route(db = Depends(get_db)): ...最灵活,可组合多个依赖

6.5 子依赖与依赖嵌套

方法名称语法用途代码示例注意事项
子依赖def dep_a(dep_b = Depends(...))依赖之间相互调用def get_token(): ... / def get_user(token = Depends(get_token)): / return user / def route(user = Depends(get_user)): ...DI 容器自动解析层级
嵌套调用A → B → C构建依赖链支持任意深度嵌套避免循环依赖
参数复用同一依赖被多个函数使用减少重复代码get_db 可被所有数据操作函数使用推荐将通用逻辑封装为依赖

6.6 使用 Depends() 传递参数与复用逻辑

方法名称语法用途代码示例注意事项
参数化依赖Depends(lambda: func(arg))向依赖传递参数@app.get("/items", dependencies=[Depends(lambda: check_role("editor"))])使用 lambda 包装
部分应用from functools import partial固定部分参数from functools import partial / editor_only = partial(check_role, role="editor") / Depends(editor_only)更清晰的参数传递方式
动态配置Depends(configured_func)根据环境或路径动态选择逻辑根据 URL 路径切换数据库实例提高灵活性和可配置性

第七章:安全与认证

7.1 OAuth2 + Password(含 Bearer)流程

概念说明用途代码示例注意事项
OAuth2PasswordBearer用于密码模式的 Bearer token 认证获取并验证 tokenfrom fastapi.security import OAuth2PasswordBearer / oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")tokenUrl 是获取 token 的端点
Bearer TokenAuthorization: Bearer <token>客户端在请求头中携带 token请求头示例:Authorization: Bearer eyJhbGciOi...传输需 HTTPS
密码模式用户名+密码换取 token实现登录认证通常用于内部系统或可信客户端不推荐用于第三方应用

7.2 使用 OAuth2PasswordRequestForm 处理登录

方法名称语法用途代码示例注意事项
OAuth2PasswordRequestFormform_data: OAuth2PasswordRequestForm = Depends()接收用户名密码表单@app.post("/token") / def login(form_data: OAuth2PasswordRequestForm = Depends()): / user = authenticate(form_data.username, form_data.password)自动解析 x-www-form-urlencoded
表单字段form_data.username, form_data.password获取认证凭据需配合密码哈希验证密码不应明文存储
安全传输HTTPS防止凭据泄露生产环境必须使用 HTTPS否则存在安全风险

7.3 JWT(JSON Web Token)生成与验证

方法名称语法用途代码示例注意事项
jwt.encode()jwt.encode(payload, key, algorithm)生成 JWT tokenimport jwt / token = jwt.encode({"sub": "user123"}, "SECRET", algorithm="HS256")使用强密钥
jwt.decode()jwt.decode(token, key, algorithms)验证并解析 tokenpayload = jwt.decode(token, "SECRET", algorithms=["HS256"])捕获 ExpiredSignatureError 等异常
载荷(Payload){"sub": "...", "exp": ...}存储用户信息和过期时间exp 字段表示过期时间戳避免存储敏感信息

7.4 使用 Security 依赖保护路由

方法名称语法用途代码示例注意事项
Security()Security(dependency, scopes=...)声明安全依赖和权限范围@app.get("/me") / def read_me(current_user = Security(get_current_user, scopes=["me"])): ...用于 OAuth2 作用域控制
权限校验raise HTTPException(403)在依赖中拒绝未授权访问if not has_scope(user, scope): / raise HTTPException(403)及时中断请求
作用域(Scopes)scopes=["read", "write"]细粒度权限控制适用于复杂权限系统需在 token 中包含 scope 信息

7.5 API Key 认证(Header、Query、Cookie)

方法名称语法用途代码示例注意事项
APIKeyHeaderAPIKeyHeader(name="X-API-Key")从请求头读取 API Keyfrom fastapi.security import APIKeyHeader / api_key_header = APIKeyHeader(name="X-API-Key")常见于第三方 API
APIKeyQueryAPIKeyQuery(name="api_key")从查询参数读取 API Keyapi_key_q = APIKeyQuery(name="api_key")URL 中暴露 key,安全性较低
APIKeyCookieAPIKeyCookie(name="session")从 Cookie 读取 API Keyapi_key_cookie = APIKeyCookie(name="session")需防范 CSRF 攻击
验证逻辑if key != VALID_KEY: raise ...校验 key 是否有效可结合数据库或缓存验证建议使用哈希存储 key

7.6 权限控制与角色管理基础

方法名称语法用途代码示例注意事项
角色检查if user.role != "admin": ...控制不同角色的访问权限def is_admin(user: User = Depends(get_current_user)): / if user.role != "admin": / raise HTTPException(403) / return user可作为依赖复用
装饰器模式@require_role("admin")简化权限检查自定义装饰器封装角色逻辑提高代码可读性
RBAC 模型Role → Permissions实现基于角色的访问控制数据库设计包含用户、角色、权限表适用于复杂系统

第八章:数据库集成(SQLAlchemy / Tortoise ORM)

8.1 同步数据库(SQLAlchemy + Session)

方法名称语法用途代码示例注意事项
SessionLocalsession = SessionLocal()创建数据库会话from sqlalchemy.orm import sessionmaker / SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)每个请求应创建独立会话
依赖注入会话def get_db(): yield session将数据库会话作为依赖def get_db(): / db = SessionLocal() / try: / yield db / finally: / db.close()使用 yield 确保关闭
CRUD 操作db.query(Model).filter(...).first()执行数据库操作user = db.query(User).filter(User.email==email).first()注意查询性能,避免 N+1 问题

8.2 异步数据库(Tortoise ORM / SQLAlchemy 2.0 async)

方法名称语法用途代码示例注意事项
async with connasync with database.connect():异步连接数据库async with get_db() as db: / users = await User.all()需使用异步 ORM
await 查询result = await query非阻塞执行数据库操作users = await db.fetch_all(users.select())保持异步上下文
异步会话async_session = AsyncSession()SQLAlchemy 2.0 异步支持from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession / engine = create_async_engine(DATABASE_URL)需安装 async driver(如 asyncpg)

8.3 使用 @app.on_event(“startup”) 初始化数据库连接

方法名称语法用途代码示例注意事项
startup 事件@app.on_event("startup")应用启动时执行初始化@app.on_event("startup") / async def init_db(): / await database.connect()适用于异步初始化
shutdown 事件@app.on_event("shutdown")应用关闭时清理资源@app.on_event("shutdown") / async def disconnect_db(): / await database.disconnect()释放连接、关闭文件等
同步初始化同步函数直接调用同步数据库连接create_tables()可在模块级执行

8.4 CRUD 操作封装与依赖注入

方法名称语法用途代码示例注意事项
CRUD 类class CRUDBase: def create(...):封装通用数据操作class CRUDUser(CRUDBase): / def get_by_email(self, db, email): ...提高代码复用性
依赖注入服务def route(service = Depends(UserService))注入业务服务类解耦路由与数据访问逻辑便于测试和替换实现
分页查询skip, limit 参数实现数据分页def get_items(db, skip=0, limit=10): / return db.query(Item).offset(skip).limit(limit).all()避免一次性加载大量数据

8.5 分页与查询优化

方法名称语法用途代码示例注意事项
OFFSET/LIMIT.offset(skip).limit(limit)实现分页items = db.query(Item).offset(10).limit(20).all()大偏移量时性能下降
游标分页WHERE id > last_id LIMIT n基于游标的高效分页items = db.query(Item).filter(Item.id > last_id).limit(20).all()适用于不可变数据集
索引优化CREATE INDEX ...提高查询速度为常用查询字段(如 email, status)创建索引避免全表扫描
懒加载 vs 预加载selectinload, joinedload控制关联对象加载策略from sqlalchemy.orm import selectinload / stmt = select(User).options(selectinload(User.items))防止 N+1 查询问题

第九章:异步编程与性能优化

9.1 async def 与 await 基础

方法名称语法用途代码示例注意事项
async defasync def func(): ...定义异步函数async def fetch_data(): / await asyncio.sleep(1) / return "data"只能在 async 函数中使用 await
await 表达式await coro()等待协程完成result = await fetch_data()释放控制权,允许其他任务运行
协程对象func() 返回协程异步调用不立即执行task = fetch_data() # coroutine需 await 或 asyncio.create_task()

9.2 异步视图函数与非阻塞 I/O

方法名称语法用途代码示例注意事项
异步路由@app.get("/") / async def handler(): ...处理异步请求@app.get("/data") / async def get_data(): / return await db.fetch_data()适用于数据库、HTTP 请求等耗时操作
非阻塞调用await asyncio.sleep(), await db.query()执行非阻塞操作避免使用 time.sleep(), requests.get() 等阻塞调用使用 aiohttp、asyncpg 等异步库

9.3 异步数据库操作示例

方法名称语法用途代码示例注意事项
异步查询result = await conn.fetch(...)获取数据users = await database.fetch_all(users.select())需在 async 函数中
异步插入await conn.execute(...)写入数据await database.execute(users.insert().values(name="John"))注意事务管理
事务处理async with conn.transaction():确保操作原子性async with database.transaction(): / await db.execute(...)防止数据不一致

9.4 并发请求处理与性能测试

方法名称语法用途代码示例注意事项
asyncio.gather()await asyncio.gather(coro1, coro2)并发执行多个协程results = await asyncio.gather(fetch_user(1), fetch_user(2))比顺序执行更快
性能测试工具pytest-benchmark, wrk测量 API 性能wrk -t10 -c100 -d30s http://localhost:8000/评估并发处理能力
压力测试模拟高并发请求发现性能瓶颈使用 Locust 或 k6关注 CPU、内存、数据库连接数

9.5 使用 BackgroundTasks 执行后台任务

方法名称语法用途代码示例注意事项
BackgroundTasksbackground_tasks: BackgroundTasks注入后台任务队列from fastapi import BackgroundTasks / def send_email(to: str): ... / @app.post("/notify", status_code=202) / def notify(bg: BackgroundTasks): / bg.add_task(send_email, "user@example.com")立即返回响应,任务后台执行
add_task()bg.add_task(func, *args, **kwargs)添加任务到队列bg.add_task(generate_report, user_id=123)任务函数不应有返回值依赖
适用场景发送邮件、生成报告、清理缓存耗时但无需即时响应的操作避免在后台任务中处理关键业务逻辑任务可能在应用关闭时中断

第十章:中间件与事件处理器

10.1 添加中间件(CORS、日志、GZip)

方法名称语法用途代码示例注意事项
CORSMiddlewareapp.add_middleware(CORSMiddleware, ...)处理跨域请求from fastapi.middleware.cors import CORSMiddleware / app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])前端开发常用
GZipMiddlewareapp.add_middleware(GZipMiddleware)启用响应压缩from starlette.middleware.gzip import GZipMiddleware / app.add_middleware(GZipMiddleware, minimum_size=1000)减少传输体积,提高性能
日志中间件自定义中间件记录请求信息监控和调试记录请求路径、方法、耗时、状态码注意不要记录敏感数据

10.2 自定义中间件(@app.middleware(“http”))

方法名称语法用途代码示例注意事项
@app.middleware(“http”)@app.middleware("http") / async def middleware(request, call_next): ...定义自定义中间件@app.middleware("http") / async def add_process_time_header(request: Request, call_next): / start_time = time.time() / response = await call_next(request) / response.headers["X-Process-Time"] = str(time.time() - start_time) / return response在请求前后执行逻辑
call_nextresponse = await call_next(request)调用下一个中间件或路由函数必须调用以继续处理流程可修改请求或响应对象

10.3 启动与关闭事件(@app.on_event(“startup”) / “shutdown”)

方法名称语法用途代码示例注意事项
@app.on_event(“startup”)@app.on_event("startup") / async def func(): ...应用启动时初始化资源连接数据库、加载模型、创建索引确保资源准备就绪
@app.on_event(“shutdown”)@app.on_event("shutdown") / async def func(): ...应用关闭时清理资源断开数据库连接、关闭文件句柄、保存状态避免资源泄漏
同步事件同步函数执行同步初始化create_tables()可直接定义函数

10.4 请求生命周期钩子

概念说明用途示例注意事项
请求开始接收到请求记录开始时间、验证身份中间件中记录 start_time可用于性能监控
路由匹配查找对应处理函数确定执行哪个路径操作函数框架自动处理注意路径顺序
依赖注入解析并执行依赖获取数据库连接、用户身份等Depends(get_current_user)可能抛出异常
请求处理执行主函数逻辑业务逻辑处理CRUD 操作、计算等应尽量异步
响应生成序列化返回值生成 JSON 或其他格式响应return {"data": ...}可通过 response_model 过滤
响应发送将响应返回客户端完成 HTTP 交互框架自动处理中间件可修改响应头

第十一章:API 文档与交互式界面

11.1 自动生成的 Swagger UI(/docs)

方法名称语法用途代码示例注意事项
Swagger UI访问 /docs交互式 API 文档界面启动应用后访问 http://localhost:8000/docs基于 Swagger UI 构建
实时测试在浏览器中直接调用 API调试和演示 API点击 “Try it out” 按钮需处理认证(如 Bearer Token)
自动更新修改代码后刷新页面文档与代码同步无需手动维护文档极大提升开发效率

11.2 ReDoc 文档界面(/redoc)

方法名称语法用途代码示例注意事项
ReDoc访问 /redoc另一种 API 文档展示形式http://localhost:8000/redoc基于 ReDoc 构建
阅读友好更适合阅读和分享技术文档、对外 API 说明页面布局更清晰不支持直接测试请求
OpenAPI 规范/openapi.json生成标准 OpenAPI(原 Swagger)JSON可用于导入 Postman、生成 SDK标准化接口描述

11.3 自定义 API 元数据(标题、描述、版本)

方法名称语法用途代码示例注意事项
FastAPI 构造函数参数FastAPI(title=..., description=..., version=...)设置 API 基本信息app = FastAPI(title="My API", description="A sample API", version="1.0.0")出现在文档首页
文档 URL 自定义docs_url, redoc_url修改或禁用文档路径app = FastAPI(docs_url="/api-docs") / # 或 app = FastAPI(docs_url=None) 禁用生产环境可禁用以提高安全
联系信息contact=...提供开发者联系方式contact={"name": "API Support", "email": "support@example.com"}便于使用者联系

11.4 给路由添加文档字符串(docstring)

方法名称语法用途代码示例注意事项
函数文档字符串"""..."""为路由添加详细说明@app.get("/items") / def read_items(): / """Retrieve a list of items. Optional query parameters: - skip: items to skip - limit: max items to return""" / return []支持 Markdown 格式
参数描述:param param: ...详细描述参数在 docstring 中使用标准格式提高文档可读性
响应描述:return: ...描述返回值文档自动生成响应模型可结合 response_model 使用

11.5 隐藏或自定义文档路径

方法名称语法用途代码示例注意事项
禁用文档docs_url=None, redoc_url=None关闭自动生成的文档app = FastAPI(docs_url=None, redoc_url=None)生产环境增强安全性
自定义路径docs_url="/apidocs"修改文档访问路径app = FastAPI(docs_url="/apidocs")避免被扫描发现
条件启用根据环境变量决定开发环境开启,生产关闭if settings.DEBUG: / app = FastAPI() / else: / app = FastAPI(docs_url=None)灵活控制

第十二章:测试与调试

12.1 使用 TestClient 进行单元测试

方法名称语法用途代码示例注意事项
TestClientfrom fastapi.testclient import TestClient创建测试客户端client = TestClient(app) / response = client.get("/")同步接口,便于测试
模拟请求client.get(), client.post()发送各种 HTTP 请求response = client.post("/items", json={"name": "foo"})可传递 json, data, headers 等参数
测试覆盖率pytest-cov测量代码测试覆盖率pytest --cov=app tests/确保关键路径被覆盖

12.2 模拟请求与断言响应

方法名称语法用途代码示例注意事项
assert response.status_code == 200验证状态码确保请求成功assert response.status_code == 200常见断言
assert response.json() == {…}验证响应内容检查返回数据正确性assert response.json() == {"msg": "ok"}自动解析 JSON
assert “required” in response.text验证文本内容检查 HTML 或文本响应适用于非 JSON 响应使用 response.text

12.3 测试依赖覆盖(app.dependency_overrides)

方法名称语法用途代码示例注意事项
dependency_overridesapp.dependency_overrides[dep] = mock替换依赖为模拟实现def override_get_db(): / return TestingSessionLocal() / app.dependency_overrides[get_db] = override_get_db隔离测试,避免真实数据库
清理覆盖app.dependency_overrides.clear()测试后恢复原始依赖在 pytest fixture teardown 中调用防止影响其他测试
模拟外部服务模拟邮件发送、第三方 API测试时不发送真实请求def mock_send_email(to, body): pass / app.dependency_overrides[send_email] = mock_send_email提高测试速度和稳定性

12.4 异步测试支持(pytest-asyncio)

方法名称语法用途代码示例注意事项
@pytest.mark.asyncio@pytest.mark.asyncio标记异步测试函数@pytest.mark.asyncio / async def test_async_route(): / response = client.get("/async") / assert response.status_code == 200需安装 pytest-asyncio
AsyncClientfrom starlette.testclient import TestClient (异步版)异步测试客户端推荐使用 httpx.AsyncClient更适合测试异步逻辑
await client.get()await 发送请求在异步测试中使用response = await async_client.get("/items")保持异步上下文

12.5 调试技巧与日志配置

方法名称语法用途代码示例注意事项
print() 调试print("debug info")快速输出变量值print(f"User: {user}")简单但有效,生产移除
logging 模块import logging标准日志记录logging.basicConfig(level=logging.INFO) / logger = logging.getLogger(__name__) / logger.info("Request processed")可配置级别、格式、输出位置
IDE 调试器断点调试逐步执行代码在 PyCharm、VSCode 中设置断点最强大的调试方式
环境区分DEBUG=True开发环境启用详细日志if settings.DEBUG: / logging.basicConfig(level=logging.DEBUG)生产环境使用 WARNING 或 ERROR

第十三章:部署与生产环境配置

13.1 使用 Uvicorn 生产模式启动

方法名称语法用途代码示例注意事项
Uvicorn 命令行uvicorn main:app --host 0.0.0.0 --port 80 --workers 4生产环境启动使用多个 worker 提升性能避免使用 —reload
workers 参数--workers N启动多个工作进程--workers 4 # 通常为 CPU 核心数提高并发处理能力
worker_class--worker-class uvicorn.workers.UvicornWorker指定工作类与 Gunicorn 配合使用优化性能

13.2 部署到 Gunicorn + Uvicorn 工作进程

方法名称语法用途代码示例注意事项
Gunicorn 命令gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app使用 Gunicorn 管理 Uvicorn 进程结合 Gunicorn 的进程管理与 Uvicorn 的异步能力生产推荐方案
配置文件gunicorn.conf.py集中管理部署配置bind = "0.0.0.0:8000" / workers = 4 / worker_class = "uvicorn.workers.UvicornWorker"便于版本控制和复用
进程监控Gunicorn 自带监控重启崩溃进程、负载均衡自动管理 worker 生命周期提高应用稳定性

13.3 使用 Docker 容器化部署

方法名称语法用途代码示例注意事项
DockerfileFROM, COPY, CMD定义应用镜像FROM python:3.9 / COPY . /app / WORKDIR /app / RUN pip install -r requirements.txt / CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]基于官方 Python 镜像
docker-compose.ymlservices: app: ...定义多容器应用包含 app、db、redis 等服务便于本地开发和部署
镜像构建docker build -t myapp .构建 Docker 镜像tag 用于版本管理推送到镜像仓库

13.4 环境变量管理(pydantic.BaseSettings)

方法名称语法用途代码示例注意事项
BaseSettingsclass Settings(BaseSettings): ...管理应用配置class Settings(BaseSettings): / database_url: str / debug: bool = False / secret_key: str / settings = Settings()自动从环境变量读取
.env 文件DATABASE_URL=sqlite:///db.sqlite3本地开发配置文件创建 .env 文件存放配置不应提交到版本控制
字段别名field_name: str = Field(..., env="ENV_NAME")映射不同环境变量名api_key: str = Field(..., env="SECRET_API_KEY")灵活适配

13.5 日志配置与监控(Prometheus、Sentry)

方法名称语法用途代码示例注意事项
Structured LoggingJSON 格式日志便于日志收集和分析{"time": "...", "level": "INFO", "message": "...", "path": "/items"}使用 loguru 或 structlog
Prometheus/metrics 端点收集应用指标请求计数、响应时间、错误率需集成 starlette_exporter
Sentrysentry_sdk.init()错误追踪与告警import sentry_sdk / sentry_sdk.init(dsn="...")捕获未处理异常,便于排查
健康检查@app.get("/healthz")监控应用健康状态return {"status": "ok"}用于 Kubernetes 等编排系统

第十四章:高级特性与扩展

14.1 WebSockets 支持(@app.websocket)

方法名称语法用途代码示例注意事项
@app.websocket()@app.websocket("/ws")定义 WebSocket 路由@app.websocket("/ws") / async def websocket_endpoint(websocket: WebSocket): / await websocket.accept() / while True: / data = await websocket.receive_text() / await websocket.send_text(f"Echo: {data}")实现双向实时通信
websocket.accept()await websocket.accept()接受客户端连接必须在接收消息前调用可指定子协议
receive_text() / send_text()await websocket.receive_text()收发文本消息await websocket.send_text("Hello")异步非阻塞
receive_json() / send_json()await websocket.receive_json()收发 JSON 消息data = await websocket.receive_json()自动序列化/反序列化

14.2 使用 APIRouter 模块化路由

方法名称语法用途代码示例注意事项
APIRouter()router = APIRouter()创建子路由组from fastapi import APIRouter / router = APIRouter(prefix="/users", tags=["users"])用于功能模块划分
include_router()app.include_router(router)将子路由挂载到主应用app.include_router(user_router) / app.include_router(item_router, prefix="/items")可设置统一前缀和标签
路由分组按功能拆分文件组织大型项目users.py, items.py, auth.py提高代码可维护性

14.3 子应用程序挂载(app.mount())

方法名称语法用途代码示例注意事项
app.mount()app.mount("/static", StaticFiles(...))挂载子应用from fastapi.staticfiles import StaticFiles / app.mount("/static", StaticFiles(directory="static"), name="static")用于静态文件服务
WSGI 应用mount(FlaskApp())集成 Flask 等 WSGI 应用可逐步迁移旧系统注意路径冲突
路径前缀mount("/sub", app)为子应用设置前缀所有子应用路由基于该前缀必须唯一

14.4 自定义响应类(JSONResponse、PlainTextResponse 等)

方法名称语法用途代码示例注意事项
JSONResponsereturn JSONResponse(content={})自定义 JSON 响应可设置 headers、status_code最常用
PlainTextResponsereturn PlainTextResponse("text")返回纯文本用于简单文本或脚本media_type=text/plain
HTMLResponsereturn HTMLResponse("<html>...")返回 HTML 页面可结合 Jinja2 模板引擎media_type=text/html
RedirectResponsereturn RedirectResponse(url)重定向status_code=302 或 307用于跳转
StreamingResponsereturn StreamingResponse(generator)流式传输大内容适用于视频、大文件下载节省内存

14.5 与 GraphQL 集成(Strawberry 或 Ariadne)

方法名称语法用途代码示例注意事项
Strawberry@strawberry.type定义 GraphQL 类型import strawberry / @strawberry.type / class User: / name: str / age: intPython 原生类型风格
Ariadnemake_executable_schema基于 Schema First定义 .graphql 文件和解析器适合团队协作
挂载 GraphQL 端点app.add_route("/graphql", GraphQLApp(...))将 GraphQL 集成到 FastAPIfrom strawberry.fastapi import GraphQLRouter / router = GraphQLRouter(schema) / app.include_router(router, prefix="/graphql")共享同一服务器
类型安全自动生成类型减少错误编辑器自动补全提高开发效率