FastAPI 从入门到实战：全面掌握 Python Web 框架

FastAPI 从入门到实战：全面掌握 Python Web 框架

推荐阅读：FastAPI文档

目录

一、FastAPI 基础入门

1.1 FastAPI 框架基础

什么是 FastAPI？

FastAPI 是一个用于构建 API 的现代、高性能 Python Web 框架，基于 Python 类型提示（Type Hints） 和 ASGI 异步标准 构建。它与 Node.js、Go 等语言的主流框架性能相当，是目前 Python 生态中最快的 Web 框架之一。

核心特性

环境搭建

# 安装 FastAPI 和 ASGI 服务器 uvicorn
pip install fastapi
pip install uvicorn[standard]

第一个 FastAPI 应用

from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义路由
@app.get("/")
async def root():
return {"message": "Hello, FastAPI!"}

@app.get("/hello/{name}")
async def hello(name: str):
return {"message": f"Hello, {name}!"}

启动服务：

uvicorn main:app –reload –host 0.0.0.0 –port 8000

参数说明：

• main:app：main.py 文件中的 app 实例
• –reload：代码变更后自动重启（开发环境使用）
• –host：监听地址
• –port：监听端口

启动后访问：

• API 接口：http://localhost:8000/
• Swagger UI 文档：http://localhost:8000/docs
• ReDoc 文档：http://localhost:8000/redoc

HTTP 请求方法

FastAPI 通过装饰器支持所有标准 HTTP 方法：

@app.get("/items") # GET – 查询数据
@app.post("/items") # POST – 创建数据
@app.put("/items/{id}") # PUT – 全量更新数据
@app.patch("/items/{id}") # PATCH – 部分更新数据
@app.delete("/items/{id}")# DELETE – 删除数据

1.2 参数分类

在 Web API 开发中，客户端向服务端传递数据的方式主要有三种：路径参数、查询参数 和 请求体参数。FastAPI 通过 Python 类型提示和 Pydantic 模型，能够自动识别并校验这些参数。

1.2.1 路径参数

路径参数是 URL 路径中的一部分，用于标识特定资源。例如 /users/123 中的 123 就是路径参数。

基本用法：

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""
user_id 是路径参数，FastAPI 会自动将 URL 中的值转换为 int 类型。
如果传入非整数（如 "abc"），会返回 422 校验错误。
"""
return {"user_id": user_id, "type": str(type(user_id))}

请求示例：

GET /users/42
响应: {"user_id": 42, "type": "<class 'int'>"}

GET /users/abc
响应: 422 Unprocessable Entity（类型校验失败）

使用 Enum 约束路径参数取值：

from enum import Enum

class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
if model_name is ModelName.alexnet:
return {"model": model_name, "message": "Deep Learning FTW!"}
if model_name.value == "resnet":
return {"model": model_name, "message": "LeCNN all the images"}
return {"model": model_name, "message": "Have some residuals"}

注意：路径参数的顺序很重要。如果同时定义了 /users/me 和 /users/{user_id}，必须将固定路径 /users/me 放在前面，否则 me 会被当作 user_id 的值。

1.2.2 路径参数 – 类型注解 Path

Path 是 FastAPI 提供的路径参数校验工具，可以添加更丰富的校验规则和元数据。

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(
..., # … 表示该参数为必填项
title="物品 ID", # 参数标题（显示在文档中）
description="要查询的物品唯一标识符", # 参数描述
ge=1, # 最小值（大于等于）
le=10000, # 最大值（小于等于）
)
):
return {"item_id": item_id}

Path 常用参数一览：

正则表达式示例：

@app.get("/files/{file_path:path}")
async def read_file(
file_path: str = Path(..., regex=r"^[\\w\\-. ]+$")
):
return {"file_path": file_path}

提示：file_path:path 中的 :path 是 Starlette 的路径转换器，允许路径参数中包含 / 字符，适用于文件路径等场景。

1.2.3 查询参数

查询参数是 URL 中 ? 后面的键值对，通常用于过滤、排序、分页等非资源标识类的数据传递。

基本用法：

from fastapi import FastAPI

app = FastAPI()

# 模拟数据库
fake_items_db = [
{"name": "iPhone", "price": 6999, "brand": "Apple"},
{"name": "Galaxy S24", "price": 5999, "brand": "Samsung"},
{"name": "Pixel 8", "price": 4999, "brand": "Google"},
{"name": "MacBook Pro", "price": 14999, "brand": "Apple"},
]

@app.get("/items")
async def list_items(brand: str | None = None, max_price: int | None = None):
"""
brand 和 max_price 都是查询参数（非路径参数），
未出现在路径中且有默认值，FastAPI 自动识别为查询参数。
"""
results = fake_items_db
if brand:
results = [item for item in results if item["brand"] == brand]
if max_price:
results = [item for item in results if item["price"] <= max_price]
return {"count": len(results), "items": results}

请求示例：

GET /items?brand=Apple&max_price=10000
响应: {
"count": 1,
"items": [{"name": "iPhone", "price": 6999, "brand": "Apple"}]
}

接收列表类型的查询参数：

from fastapi import Query

@app.get("/items")
async def list_items(tags: list[str] = Query(default=[])):
"""接收多个 tag 值，如 /items?tags=electronics&tags=new"""
return {"tags": tags}

1.2.4 查询参数 – 类型注解 Query

与 Path 类似，Query 提供了对查询参数的增强校验能力。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/search")
async def search_items(
q: str = Query(
default=None, # 默认值（None 表示可选参数）
title="搜索关键词",
description="用于搜索物品名称的关键词",
min_length=2, # 最少 2 个字符
max_length=50, # 最多 50 个字符
),
page: int = Query(
default=1,
ge=1,
description="页码，从 1 开始"
),
size: int = Query(
default=10,
ge=1,
le=100,
description="每页数量，1-100"
),
):
return {"query": q, "page": page, "size": size}

请求示例：

GET /search?q=iPhone&page=2&size=5
响应: {"query": "iPhone", "page": 2, "size": 5}

GET /search?q=a
响应: 422（q 最少需要 2 个字符）

参数别名（alias）：

@app.get("/items")
async def list_items(
# 前端传参使用 item-size，Python 代码中使用 item_size
item_size: int = Query(default=10, alias="item-size")
):
return {"size": item_size}

Path vs Query 总结：路径参数用于标识资源（如用户 ID），查询参数用于过滤/分页。两者都支持丰富的校验规则，区别仅在于参数来源不同。

1.2.5 请求体参数

请求体参数通过 HTTP 请求体（Body）传递数据，通常用于 POST、PUT、PATCH 等需要提交复杂数据的场景。FastAPI 使用 Pydantic 模型 来定义和校验请求体数据。

基本用法：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义 Pydantic 模型
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None

@app.post("/items")
async def create_item(item: Item):
"""
item 参数类型为 Pydantic 模型，
FastAPI 会自动：
1. 读取请求体 JSON
2. 校验数据类型和约束
3. 将数据转换为 Item 实例
"""
item_dict = item.model_dump() # 转换为字典
if item.tax:
price_with_tax = item.price + item.tax
item_dict["price_with_tax"] = price_with_tax
return item_dict

请求示例：

POST /items
Content–Type: application/json

{
"name": "iPhone 16",
"description": "Apple 最新旗舰手机",
"price": 7999.0,
"tax": 1.06
}

响应: {
"name": "iPhone 16",
"description": "Apple 最新旗舰手机",
"price": 7999.0,
"tax": 1.06,
"price_with_tax": 8479.94
}

嵌套模型与复杂结构：

from pydantic import BaseModel

class Image(BaseModel):
url: str
name: str

class User(BaseModel):
username: str
full_name: str | None = None
age: int
tags: list[str] = []
images: list[Image] = [] # 嵌套模型

@app.post("/users")
async def create_user(user: User):
return user

请求体示例：

{
"username": "zhangsan",
"full_name": "张三",
"age": 28,
"tags": ["developer", "python"],
"images": [
{"url": "https://example.com/avatar.jpg", "name": "avatar"}
]
}

1.2.6 请求体参数 – 类型注解 Field

Field 用于对 Pydantic 模型中的字段添加额外的校验规则和元数据信息，类似于 Path 和 Query 的作用。

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Product(BaseModel):
name: str = Field(
..., # 必填
title="产品名称",
description="产品的完整名称",
min_length=1,
max_length=100,
examples=["iPhone 16 Pro Max"]
)
price: float = Field(
...,
title="价格",
description="产品售价（单位：元）",
gt=0, # 必须大于 0
le=999999, # 不超过 999999
)
stock: int = Field(
default=0,
title="库存数量",
ge=0, # 不能为负数
)
category: str = Field(
default="未分类",
description="产品分类"
)

@app.post("/products")
async def create_product(product: Product):
return {
"message": "产品创建成功",
"product": product.model_dump()
}

Field 与 JSON Schema：

Field 中的所有元数据（title、description、examples 等）会自动反映在 Swagger UI 和 ReDoc 文档中，极大提升了 API 文档的质量和可维护性。

访问 http://localhost:8000/docs 可以看到：
– 每个字段都有中文标题和描述
– 字段约束（范围、长度等）清晰标注
– 提供了 example 值供快速测试

三种参数小结

1.3 响应类型

FastAPI 支持多种响应格式，默认返回 JSON，也可以灵活返回 HTML、文件等。

1.3.1 响应类型 – JSON 格式

JSON 是 FastAPI 的默认响应格式，直接返回 dict、list 或 Pydantic 模型即可。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ItemResponse(BaseModel):
id: int
name: str
price: float

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
# response_model 会自动过滤多余字段，并生成文档
raw_data = {"id": item_id, "name": "Test", "price": 99.9, "internal_code": "SECRET"}
return raw_data # 响应中不会包含 internal_code

使用 response_model 的优势：

• 自动过滤：只返回模型中定义的字段，隐藏敏感信息
• 类型转换：自动将返回值转换为指定类型
• 文档生成：Swagger UI 中自动展示响应结构

多状态码响应：

from fastapi.responses import JSONResponse

@app.post("/items", status_code=201)
async def create_item():
return {"message": "Item created"}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
# 返回 204 No Content
return JSONResponse(status_code=204, content=None)

1.3.2 响应类型设置方式

FastAPI 提供了多种设置响应类型的方式：

方式一：通过 response_class 参数

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, PlainTextResponse

app = FastAPI()

@app.get("/html", response_class=HTMLResponse)
async def get_html():
return "<h1>Hello, HTML!</h1>"

@app.get("/text", response_class=PlainTextResponse)
async def get_text():
return "纯文本响应"

方式二：直接返回 Response 对象（更灵活）

from fastapi import FastAPI
from fastapi.responses import JSONResponse, RedirectResponse

app = FastAPI()

@app.get("/json-custom")
async def custom_json():
# 手动设置状态码和响应头
return JSONResponse(
status_code=200,
content={"message": "自定义 JSON"},
headers={"X-Custom-Header": "MyValue"}
)

@app.get("/redirect")
async def redirect():
return RedirectResponse(url="/docs")

方式三：通过 response_model + 路由装饰器

@app.get("/items/{item_id}", response_model=ItemResponse, status_code=200)
async def get_item(item_id: int):
...

1.3.3 响应 HTML 格式

使用 HTMLResponse 返回 HTML 页面内容。

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi import Request

app = FastAPI()

# 方式一：直接返回 HTML 字符串
@app.get("/hello-html", response_class=HTMLResponse)
async def hello_html():
return """
<!DOCTYPE html>
<html>
<head><title>FastAPI HTML</title></head>
<body>
<h1 style="color: blue;">Hello from FastAPI!</h1>
<p>这是一个 HTML 响应示例。</p>
</body>
</html>
"""

# 方式二：使用 Jinja2 模板引擎（推荐）
# pip install jinja2
templates = Jinja2Templates(directory="templates")

@app.get("/page/{name}", response_class=HTMLResponse)
async def render_page(request: Request, name: str):
return templates.TemplateResponse(
"index.html",
{"request": request, "name": name}
)

模板文件 templates/index.html：

<!DOCTYPE html>
<html>
<head><title>{{ name }} 的主页</title></head>
<body>
<h1>欢迎你，{{ name }}！</h1>
</body>
</html>

1.3.4 响应文件格式

使用 FileResponse 和 StreamingResponse 返回文件内容。

from fastapi import FastAPI
from fastapi.responses import FileResponse, StreamingResponse
import io

app = FastAPI()

# 返回静态文件
@app.get("/download/report")
async def download_report():
return FileResponse(
path="reports/monthly_report.pdf",
filename="月度报告.pdf", # 下载时的文件名
media_type="application/pdf", # MIME 类型
)

# 返回图片
@app.get("/avatar/{user_id}")
async def get_avatar(user_id: int):
return FileResponse(
path=f"avatars/{user_id}.png",
media_type="image/png"
)

# 流式响应（适合大文件或动态生成的内容）
@app.get("/stream")
async def stream_data():
def generate():
for i in range(5):
yield f"data chunk {i}\\n"
return StreamingResponse(
io.BytesIO(b"file content here"),
media_type="application/octet-stream"
)

常用 MIME 类型参考：

1.3.5 自定义响应数据格式

在实际项目中，通常需要统一 API 的响应格式，便于前端统一处理。

定义统一响应模型：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import TypeVar, Generic

app = FastAPI()

T = TypeVar("T")

class ApiResponse(BaseModel):
code: int = 200 # 业务状态码
message: str = "success" # 提示信息
data: dict | list | None = None # 响应数据

# 使用统一格式
@app.get("/items/{item_id}", response_model=ApiResponse)
async def get_item(item_id: int):
item = {"id": item_id, "name": "iPhone", "price": 6999}
return ApiResponse(data=item)

@app.post("/items", response_model=ApiResponse, status_code=201)
async def create_item(name: str, price: float):
item = {"id": 1, "name": name, "price": price}
return ApiResponse(code=201, message="创建成功", data=item)

@app.delete("/items/{item_id}", response_model=ApiResponse)
async def delete_item(item_id: int):
return ApiResponse(message=f"物品 {item_id} 已删除")

统一响应格式示例：

// 成功响应
{
"code": 200,
"message": "success",
"data": {"id": 1, "name": "iPhone", "price": 6999}
}

// 错误响应（通过异常处理返回）
{
"code": 404,
"message": "物品不存在",
"data": null
}

1.4 异常处理

FastAPI 提供了灵活的异常处理机制，可以捕获并自定义错误响应。

内置异常 HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail=f"物品 {item_id} 不存在",
headers={"X-Error": "Item not found"}, # 自定义响应头
)
return {"item": items[item_id]}

自定义异常与异常处理器

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 1. 定义自定义异常类
class BusinessError(Exception):
def __init__(self, code: int, message: str):
self.code = code
self.message = message

class ValidationError(Exception):
def __init__(self, field: str, message: str):
self.field = field
self.message = message

# 2. 注册异常处理器
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
return JSONResponse(
status_code=200, # HTTP 状态码仍为 200，业务码在 body 中
content={
"code": exc.code,
"message": exc.message,
"data": None
}
)

@app.exception_handler(ValidationError)
async def validation_error_handler(request: Request, exc: ValidationError):
return JSONResponse(
status_code=422,
content={
"code": 422,
"message": f"字段校验失败：{exc.field} – {exc.message}",
"data": None
}
)

# 3. 在路由中抛出异常
@app.get("/pay")
async def pay(amount: float):
if amount <= 0:
raise BusinessError(code=4001, message="支付金额必须大于 0")
if amount > 10000:
raise ValidationError(field="amount", message="单笔支付不能超过 10000")
return {"message": f"支付成功，金额：{amount}"}

覆盖默认异常处理器

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

# 覆盖 422 校验错误（如参数类型不匹配）
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={
"code": 422,
"message": "请求参数校验失败",
"errors": exc.errors(), # 详细的校验错误信息
"data": None
}
)

# 覆盖所有 HTTP 异常
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"code": exc.status_code,
"message": exc.detail,
"data": None
}
)

最佳实践：在项目中统一使用自定义异常 + 全局异常处理器，确保所有错误都返回统一的 JSON 格式，便于前端统一处理。

二、FastAPI 进阶

2.1 中间件

中间件（Middleware）是在请求到达路由处理函数之前和响应返回客户端之前执行的代码。它适用于日志记录、跨域处理、认证鉴权、请求耗时统计等横切关注点。

内置中间件 – CORS

跨域资源共享（CORS）是 Web 开发中最常用的中间件场景：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 添加 CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://yourdomain.com"], # 允许的前端域名
allow_credentials=True, # 允许携带 Cookie
allow_methods=["*"], # 允许所有 HTTP 方法
allow_headers=["*"], # 允许所有请求头
)

自定义中间件

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
"""
自定义中间件：记录每个请求的处理耗时，并添加到响应头中。

执行流程：
1. 请求进入 → 记录开始时间
2. call_next(request) → 执行后续中间件和路由处理函数
3. 响应返回 → 计算耗时，添加到响应头
"""
start_time = time.perf_counter()

response = await call_next(request)

process_time = time.perf_counter() – start_time
response.headers["X-Process-Time"] = f"{process_time:.4f}s"

# 打印请求日志
print(f"[{request.method}] {request.url.path} – {process_time:.4f}s")

return response

# 日志中间件示例
@app.middleware("http")
async def log_requests(request: Request, call_next):
"""记录所有请求的详细信息"""
print(f"→ 请求: {request.method} {request.url}")
print(f" 客户端: {request.client.host}:{request.client.port}")

response = await call_next(request)

print(f"← 响应: {response.status_code}")
return response

中间件执行顺序：

请求 → 中间件A（先注册后执行） → 中间件B（后注册先执行） → 路由处理函数
响应 → 路由处理函数 → 中间件B → 中间件A → 客户端

中间件执行流程图：

#mermaid-svg-Od36cwiXYFCzYB7j{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-Od36cwiXYFCzYB7j .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-Od36cwiXYFCzYB7j .error-icon{fill:#552222;}#mermaid-svg-Od36cwiXYFCzYB7j .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-Od36cwiXYFCzYB7j .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-Od36cwiXYFCzYB7j .marker{fill:#333333;stroke:#333333;}#mermaid-svg-Od36cwiXYFCzYB7j .marker.cross{stroke:#333333;}#mermaid-svg-Od36cwiXYFCzYB7j svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-Od36cwiXYFCzYB7j p{margin:0;}#mermaid-svg-Od36cwiXYFCzYB7j .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-Od36cwiXYFCzYB7j .cluster-label text{fill:#333;}#mermaid-svg-Od36cwiXYFCzYB7j .cluster-label span{color:#333;}#mermaid-svg-Od36cwiXYFCzYB7j .cluster-label span p{background-color:transparent;}#mermaid-svg-Od36cwiXYFCzYB7j .label text,#mermaid-svg-Od36cwiXYFCzYB7j span{fill:#333;color:#333;}#mermaid-svg-Od36cwiXYFCzYB7j .node rect,#mermaid-svg-Od36cwiXYFCzYB7j .node circle,#mermaid-svg-Od36cwiXYFCzYB7j .node ellipse,#mermaid-svg-Od36cwiXYFCzYB7j .node polygon,#mermaid-svg-Od36cwiXYFCzYB7j .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-Od36cwiXYFCzYB7j .rough-node .label text,#mermaid-svg-Od36cwiXYFCzYB7j .node .label text,#mermaid-svg-Od36cwiXYFCzYB7j .image-shape .label,#mermaid-svg-Od36cwiXYFCzYB7j .icon-shape .label{text-anchor:middle;}#mermaid-svg-Od36cwiXYFCzYB7j .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-Od36cwiXYFCzYB7j .rough-node .label,#mermaid-svg-Od36cwiXYFCzYB7j .node .label,#mermaid-svg-Od36cwiXYFCzYB7j .image-shape .label,#mermaid-svg-Od36cwiXYFCzYB7j .icon-shape .label{text-align:center;}#mermaid-svg-Od36cwiXYFCzYB7j .node.clickable{cursor:pointer;}#mermaid-svg-Od36cwiXYFCzYB7j .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-Od36cwiXYFCzYB7j .arrowheadPath{fill:#333333;}#mermaid-svg-Od36cwiXYFCzYB7j .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-Od36cwiXYFCzYB7j .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-Od36cwiXYFCzYB7j .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-Od36cwiXYFCzYB7j .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-Od36cwiXYFCzYB7j .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-Od36cwiXYFCzYB7j .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-Od36cwiXYFCzYB7j .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-Od36cwiXYFCzYB7j .cluster text{fill:#333;}#mermaid-svg-Od36cwiXYFCzYB7j .cluster span{color:#333;}#mermaid-svg-Od36cwiXYFCzYB7j div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-Od36cwiXYFCzYB7j .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-Od36cwiXYFCzYB7j rect.text{fill:none;stroke-width:0;}#mermaid-svg-Od36cwiXYFCzYB7j .icon-shape,#mermaid-svg-Od36cwiXYFCzYB7j .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-Od36cwiXYFCzYB7j .icon-shape p,#mermaid-svg-Od36cwiXYFCzYB7j .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-Od36cwiXYFCzYB7j .icon-shape .label rect,#mermaid-svg-Od36cwiXYFCzYB7j .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-Od36cwiXYFCzYB7j .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-Od36cwiXYFCzYB7j .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-Od36cwiXYFCzYB7j :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

客户端请求

中间件B – 后注册先执行

中间件A – 先注册后执行

路由处理函数

构建响应

中间件A – 处理后返回

中间件B – 处理后返回

响应返回客户端

注意：FastAPI 中后注册的中间件会先执行（栈结构）。add_middleware 的顺序决定了执行的先后。

2.2 依赖注入

依赖注入（Dependency Injection）是 FastAPI 的核心设计模式之一，它允许你将通用逻辑抽取为独立函数，在需要的路由中声明依赖，由框架自动调用并注入结果。

使用依赖注入系统来共享通用逻辑，减少代码重复

没有依赖注入时的问题：

# 每个路由都要重复写 token 校验逻辑
@app.get("/users")
async def get_users(token: str):
if token != "secret":
raise HTTPException(status_code=403, detail="未授权")
return {"users": []}

@app.get("/orders")
async def get_orders(token: str):
if token != "secret": # 重复代码！
raise HTTPException(status_code=403, detail="未授权")
return {"orders": []}

使用依赖注入重构：

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

# 1. 定义依赖函数
async def verify_token(x_token: str = Header(...)):
"""
从请求头中读取 X-Token 进行校验。
如果校验失败，直接抛出异常，请求不会到达路由函数。
"""
if x_token != "my-secret-token":
raise HTTPException(status_code=403, detail="无效的 Token")
return x_token

# 2. 在路由中声明依赖
@app.get("/users")
async def get_users(token: str = Depends(verify_token)):
return {"users": ["Alice", "Bob"], "token": token}

@app.get("/orders")
async def get_orders(token: str = Depends(verify_token)):
return {"orders": ["Order-001", "Order-002"]}

带参数的依赖（使用工厂函数）：

from fastapi import Query

# 分页依赖
def pagination_params(
page: int = Query(default=1, ge=1, description="页码"),
size: int = Query(default=10, ge=1, le=100, description="每页数量"),
):
return {"skip": (page – 1) * size, "limit": size}

@app.get("/items")
async def list_items(paging: dict = Depends(pagination_params)):
# paging = {"skip": 0, "limit": 10}
return {"paging": paging, "items": [...]}

@app.get("/users")
async def list_users(paging: dict = Depends(pagination_params)):
return {"paging": paging, "users": [...]}

依赖注入应用场景

实战：数据库会话依赖

from sqlalchemy.orm import Session

# 依赖函数：使用 yield 实现自动资源管理
def get_db():
"""
yield 之前的代码：请求开始时执行（创建资源）
yield 之后的代码：请求结束时执行（释放资源）
"""
db = SessionLocal()
try:
yield db
finally:
db.close()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user

全局依赖（应用到所有路由）：

# 方式一：在 FastAPI 实例上设置全局依赖
app = FastAPI(dependencies=[Depends(verify_token)])

# 方式二：在路由组（APIRouter）上设置
from fastapi import APIRouter

router = APIRouter(dependencies=[Depends(verify_token)])

@router.get("/items")
async def get_items():
return {"items": []}

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

依赖的嵌套：

# 依赖之间可以互相依赖，形成依赖链
async def get_current_user(token: str = Depends(verify_token)):
"""依赖 verify_token 的结果"""
user = decode_token(token)
if not user:
raise HTTPException(status_code=401, detail="用户不存在")
return user

async def get_admin_user(user: dict = Depends(get_current_user)):
"""依赖 get_current_user 的结果"""
if user.get("role") != "admin":
raise HTTPException(status_code=403, detail="权限不足")
return user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, admin: dict = Depends(get_admin_user)):
# 执行到这里时，已经完成了：Token校验 → 用户认证 → 管理员权限校验
return {"message": f"用户 {user_id} 已删除"}

依赖注入执行链路图：

#mermaid-svg-P6seSachew5wTGbl{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-P6seSachew5wTGbl .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-P6seSachew5wTGbl .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-P6seSachew5wTGbl .error-icon{fill:#552222;}#mermaid-svg-P6seSachew5wTGbl .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-P6seSachew5wTGbl .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-P6seSachew5wTGbl .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-P6seSachew5wTGbl .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-P6seSachew5wTGbl .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-P6seSachew5wTGbl .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-P6seSachew5wTGbl .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-P6seSachew5wTGbl .marker{fill:#333333;stroke:#333333;}#mermaid-svg-P6seSachew5wTGbl .marker.cross{stroke:#333333;}#mermaid-svg-P6seSachew5wTGbl svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-P6seSachew5wTGbl p{margin:0;}#mermaid-svg-P6seSachew5wTGbl .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-P6seSachew5wTGbl .cluster-label text{fill:#333;}#mermaid-svg-P6seSachew5wTGbl .cluster-label span{color:#333;}#mermaid-svg-P6seSachew5wTGbl .cluster-label span p{background-color:transparent;}#mermaid-svg-P6seSachew5wTGbl .label text,#mermaid-svg-P6seSachew5wTGbl span{fill:#333;color:#333;}#mermaid-svg-P6seSachew5wTGbl .node rect,#mermaid-svg-P6seSachew5wTGbl .node circle,#mermaid-svg-P6seSachew5wTGbl .node ellipse,#mermaid-svg-P6seSachew5wTGbl .node polygon,#mermaid-svg-P6seSachew5wTGbl .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-P6seSachew5wTGbl .rough-node .label text,#mermaid-svg-P6seSachew5wTGbl .node .label text,#mermaid-svg-P6seSachew5wTGbl .image-shape .label,#mermaid-svg-P6seSachew5wTGbl .icon-shape .label{text-anchor:middle;}#mermaid-svg-P6seSachew5wTGbl .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-P6seSachew5wTGbl .rough-node .label,#mermaid-svg-P6seSachew5wTGbl .node .label,#mermaid-svg-P6seSachew5wTGbl .image-shape .label,#mermaid-svg-P6seSachew5wTGbl .icon-shape .label{text-align:center;}#mermaid-svg-P6seSachew5wTGbl .node.clickable{cursor:pointer;}#mermaid-svg-P6seSachew5wTGbl .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-P6seSachew5wTGbl .arrowheadPath{fill:#333333;}#mermaid-svg-P6seSachew5wTGbl .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-P6seSachew5wTGbl .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-P6seSachew5wTGbl .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-P6seSachew5wTGbl .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-P6seSachew5wTGbl .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-P6seSachew5wTGbl .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-P6seSachew5wTGbl .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-P6seSachew5wTGbl .cluster text{fill:#333;}#mermaid-svg-P6seSachew5wTGbl .cluster span{color:#333;}#mermaid-svg-P6seSachew5wTGbl div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-P6seSachew5wTGbl .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-P6seSachew5wTGbl rect.text{fill:none;stroke-width:0;}#mermaid-svg-P6seSachew5wTGbl .icon-shape,#mermaid-svg-P6seSachew5wTGbl .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-P6seSachew5wTGbl .icon-shape p,#mermaid-svg-P6seSachew5wTGbl .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-P6seSachew5wTGbl .icon-shape .label rect,#mermaid-svg-P6seSachew5wTGbl .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-P6seSachew5wTGbl .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-P6seSachew5wTGbl .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-P6seSachew5wTGbl :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

Token有效

用户存在

角色=admin

Token无效

用户不存在

权限不足

客户端请求

verify_token

get_current_user

get_admin_user

路由处理函数

403 Forbidden

401 Unauthorized

403 Forbidden

返回响应

上图展示了 delete_user 接口的完整依赖链：请求依次经过 Token 校验 → 用户认证 → 管理员权限校验，任一环节失败都会直接返回错误响应。

依赖注入的优势：代码复用、逻辑解耦、易于测试（可以 mock 依赖）、自动文档生成。

2.3 ORM 简介

ORM（Object-Relational Mapping，对象关系映射）是一种将数据库表映射为 Python 类的技术，让开发者可以用操作对象的方式来操作数据库，而无需直接编写 SQL 语句。

ORM 分类

Python 生态中常见的 ORM 框架：

本文以 SQLAlchemy + FastAPI 为例进行讲解。

ORM 使用流程

安装依赖 → 创建数据库引擎 → 定义模型类 → 创建数据库表 → 在路由中使用 ORM 操作数据库

ORM 请求处理完整流程图：

#mermaid-svg-ve9BSuw54Ybnm7kZ{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-ve9BSuw54Ybnm7kZ .error-icon{fill:#552222;}#mermaid-svg-ve9BSuw54Ybnm7kZ .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-ve9BSuw54Ybnm7kZ .marker{fill:#333333;stroke:#333333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .marker.cross{stroke:#333333;}#mermaid-svg-ve9BSuw54Ybnm7kZ svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-ve9BSuw54Ybnm7kZ p{margin:0;}#mermaid-svg-ve9BSuw54Ybnm7kZ .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster-label text{fill:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster-label span{color:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster-label span p{background-color:transparent;}#mermaid-svg-ve9BSuw54Ybnm7kZ .label text,#mermaid-svg-ve9BSuw54Ybnm7kZ span{fill:#333;color:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .node rect,#mermaid-svg-ve9BSuw54Ybnm7kZ .node circle,#mermaid-svg-ve9BSuw54Ybnm7kZ .node ellipse,#mermaid-svg-ve9BSuw54Ybnm7kZ .node polygon,#mermaid-svg-ve9BSuw54Ybnm7kZ .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .rough-node .label text,#mermaid-svg-ve9BSuw54Ybnm7kZ .node .label text,#mermaid-svg-ve9BSuw54Ybnm7kZ .image-shape .label,#mermaid-svg-ve9BSuw54Ybnm7kZ .icon-shape .label{text-anchor:middle;}#mermaid-svg-ve9BSuw54Ybnm7kZ .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .rough-node .label,#mermaid-svg-ve9BSuw54Ybnm7kZ .node .label,#mermaid-svg-ve9BSuw54Ybnm7kZ .image-shape .label,#mermaid-svg-ve9BSuw54Ybnm7kZ .icon-shape .label{text-align:center;}#mermaid-svg-ve9BSuw54Ybnm7kZ .node.clickable{cursor:pointer;}#mermaid-svg-ve9BSuw54Ybnm7kZ .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .arrowheadPath{fill:#333333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-ve9BSuw54Ybnm7kZ .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-ve9BSuw54Ybnm7kZ .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-ve9BSuw54Ybnm7kZ .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster text{fill:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ .cluster span{color:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-ve9BSuw54Ybnm7kZ .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-ve9BSuw54Ybnm7kZ rect.text{fill:none;stroke-width:0;}#mermaid-svg-ve9BSuw54Ybnm7kZ .icon-shape,#mermaid-svg-ve9BSuw54Ybnm7kZ .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-ve9BSuw54Ybnm7kZ .icon-shape p,#mermaid-svg-ve9BSuw54Ybnm7kZ .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-ve9BSuw54Ybnm7kZ .icon-shape .label rect,#mermaid-svg-ve9BSuw54Ybnm7kZ .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-ve9BSuw54Ybnm7kZ .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-ve9BSuw54Ybnm7kZ .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-ve9BSuw54Ybnm7kZ :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

查询/新增/更新/删除

客户端请求

FastAPI 路由匹配

依赖注入 get_db

创建数据库会话 Session

路由处理函数

Pydantic 模型校验请求参数

ORM 操作数据库

MySQL 数据库

ORM 返回结果

Pydantic 模型序列化响应

关闭数据库会话

返回 JSON 响应给客户端

上图展示了一次完整的 ORM 请求处理流程：从客户端请求到路由匹配，经过依赖注入获取数据库会话，通过 Pydantic 校验参数，ORM 操作数据库，最后序列化响应并释放资源。

下面逐步演示完整流程。

ORM – 建表

首先准备数据库。本示例使用 MySQL（也可替换为 SQLite 用于学习测试）。

— 创建数据库
CREATE DATABASE fastapi_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

ORM – 创建数据库引擎

创建 database.py 文件：

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 数据库连接字符串
# 格式: mysql+pymysql://用户名:密码@主机:端口/数据库名
DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi_demo"

# 如果使用 SQLite（无需安装 MySQL）：
# DATABASE_URL = "sqlite:///./fastapi_demo.db"

# 创建数据库引擎
engine = create_engine(
DATABASE_URL,
echo=False, # True 时打印 SQL 语句（调试用）
pool_size=10, # 连接池大小（SQLite 不支持此参数）
max_overflow=20, # 连接池最大溢出数
pool_recycle=3600, # 连接回收时间（秒）
)

# 创建会话工厂
SessionLocal = sessionmaker(
autocommit=False, # 不自动提交（手动控制事务）
autoflush=False, # 不自动刷新（手动 flush）
bind=engine
)

# 模型基类（所有模型类都继承它）
Base = declarative_base()

安装依赖：

pip install sqlalchemy pymysql

ORM – 定义模型类

创建 models.py 文件：

# models.py
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean
from sqlalchemy.sql import func
from database import Base

class User(Base):
__tablename__ = "users" # 数据库表名

id = Column(Integer, primary_key=True, index=True, autoincrement=True)
username = Column(String(50), unique=True, nullable=False, index=True, comment="用户名")
email = Column(String(100), unique=True, nullable=False, comment="邮箱")
hashed_password = Column(String(128), nullable=False, comment="加密密码")
is_active = Column(Boolean, default=True, comment="是否启用")
created_at = Column(DateTime(timezone=True), server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), comment="更新时间")

class Item(Base):
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True, autoincrement=True)
title = Column(String(100), nullable=False, index=True, comment="物品标题")
description = Column(String(500), nullable=True, comment="物品描述")
price = Column(Float, nullable=False, comment="价格")
owner_id = Column(Integer, comment="所属用户 ID") # 外键在实际项目中应使用 ForeignKey
created_at = Column(DateTime(timezone=True), server_default=func.now(), comment="创建时间")

同时定义 Pydantic 模型（用于请求/响应校验），创建 schemas.py：

# schemas.py
from pydantic import BaseModel
from datetime import datetime

# ———- User 相关 ———-
class UserBase(BaseModel):
username: str
email: str

class UserCreate(UserBase):
"""创建用户时的请求模型"""
password: str

class UserResponse(UserBase):
"""用户响应模型"""
id: int
is_active: bool
created_at: datetime

model_config = {"from_attributes": True} # 支持从 ORM 模型直接转换

# ———- Item 相关 ———-
class ItemBase(BaseModel):
title: str
description: str | None = None
price: float

class ItemCreate(ItemBase):
pass

class ItemResponse(ItemBase):
id: int
owner_id: int | None = None
created_at: datetime

model_config = {"from_attributes": True}

ORM – 创建数据库表

创建 init_db.py：

# init_db.py
from database import engine, Base

def init_database():
"""根据模型类自动创建所有数据库表"""
Base.metadata.create_all(bind=engine)
print("数据库表创建成功！")

if __name__ == "__main__":
init_database()

运行：

python init_db.py

注意：create_all 只会创建不存在的表，不会修改已存在的表结构。生产环境中建议使用 Alembic 进行数据库迁移管理。

ORM – 路由匹配中使用 ORM

创建 main.py，整合所有内容：

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine, Base
import models
import schemas

# 创建表
Base.metadata.create_all(bind=engine)

app = FastAPI(title="FastAPI ORM 实战", version="1.0.0")

# 数据库会话依赖
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

2.4 数据库操作 – 查询

查询是最常用的数据库操作，SQLAlchemy 提供了丰富的查询 API。

# —- 查询全部 —-
@app.get("/items", response_model=list[schemas.ItemResponse])
async def get_items(db: Session = Depends(get_db)):
"""查询所有物品"""
items = db.query(models.Item).all()
return items

# —- 按 ID 查询 —-
@app.get("/items/{item_id}", response_model=schemas.ItemResponse)
async def get_item(item_id: int, db: Session = Depends(get_db)):
"""根据 ID 查询单个物品"""
item = db.query(models.Item).filter(models.Item.id == item_id).first()
if not item:
raise HTTPException(status_code=404, detail="物品不存在")
return item

数据库操作 – 查询条件

from sqlalchemy import or_

# —- 条件查询 —-
@app.get("/items/search")
async def search_items(
title: str | None = None,
min_price: float | None = None,
max_price: float | None = None,
db: Session = Depends(get_db),
):
"""多条件组合查询"""
query = db.query(models.Item)

# 动态拼接查询条件
if title:
query = query.filter(models.Item.title.contains(title)) # 模糊匹配
if min_price is not None:
query = query.filter(models.Item.price >= min_price)
if max_price is not None:
query = query.filter(models.Item.price <= max_price)

items = query.all()
return {"count": len(items), "items": items}

# —- 排序与分页 —-
@app.get("/items/page/{page}")
async def get_items_paged(
page: int = 1,
size: int = 10,
sort_by: str = "created_at",
db: Session = Depends(get_db),
):
"""分页查询"""
offset = (page – 1) * size

# 排序（默认按创建时间倒序）
order_column = getattr(models.Item, sort_by, models.Item.created_at)

items = (
db.query(models.Item)
.order_by(order_column.desc())
.offset(offset)
.limit(size)
.all()
)

total = db.query(models.Item).count()

return {
"items": items,
"total": total,
"page": page,
"size": size,
"total_pages": (total + size – 1) // size,
}

常用查询方法速查：

# 精确匹配
db.query(User).filter(User.username == "zhangsan").first()

# 模糊匹配（LIKE）
db.query(User).filter(User.username.contains("zhang")).all()
db.query(User).filter(User.username.startswith("z")).all()
db.query(User).filter(User.username.endswith("n")).all()

# 范围查询
db.query(Item).filter(Item.price.between(100, 500)).all()
db.query(User).filter(User.id.in_([1, 2, 3])).all()

# 多条件（AND / OR）
db.query(User).filter(User.is_active == True, User.age > 18).all()
db.query(User).filter(or_(User.username == "a", User.username == "b")).all()

# NULL 判断
db.query(Item).filter(Item.description.is_(None)).all()
db.query(Item).filter(Item.description.isnot(None)).all()

# 聚合查询
from sqlalchemy import func
total_price = db.query(func.sum(Item.price)).scalar()
avg_price = db.query(func.avg(Item.price)).scalar()
count = db.query(func.count(Item.id)).scalar()

数据库操作 – 新增

@app.post("/items", response_model=schemas.ItemResponse, status_code=201)
async def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
"""新增物品"""
# 1. 将 Pydantic 模型转换为 ORM 模型
db_item = models.Item(
title=item.title,
description=item.description,
price=item.price,
)

# 2. 添加到会话并提交
db.add(db_item)
db.commit()

# 3. 刷新以获取数据库自动生成的字段（如 id、created_at）
db.refresh(db_item)

return db_item

# —- 批量新增 —-
@app.post("/items/batch", status_code=201)
async def create_items_batch(
items: list[schemas.ItemCreate],
db: Session = Depends(get_db),
):
"""批量创建物品"""
db_items = [
models.Item(title=item.title, description=item.description, price=item.price)
for item in items
]
db.add_all(db_items)
db.commit()

return {"message": f"成功创建 {len(db_items)} 个物品"}

新增操作的关键步骤：

1. 创建 ORM 模型实例 → models.Item(…)
2. 添加到数据库会话 → db.add(item)
3. 提交事务 → db.commit()
4. 刷新获取生成字段 → db.refresh(item)

数据库操作 – 更新

@app.put("/items/{item_id}", response_model=schemas.ItemResponse)
async def update_item(
item_id: int,
item_update: schemas.ItemCreate,
db: Session = Depends(get_db),
):
"""全量更新物品信息"""
# 1. 查询目标记录
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="物品不存在")

# 2. 更新字段
db_item.title = item_update.title
db_item.description = item_update.description
db_item.price = item_update.price

# 3. 提交事务
db.commit()
db.refresh(db_item)

return db_item

# —- 部分更新（PATCH） —-
@app.patch("/items/{item_id}")
async def partial_update_item(
item_id: int,
title: str | None = None,
price: float | None = None,
db: Session = Depends(get_db),
):
"""部分更新：只更新传入的字段"""
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="物品不存在")

# 只更新非 None 的字段
update_data = {}
if title is not None:
db_item.title = title
update_data["title"] = title
if price is not None:
db_item.price = price
update_data["price"] = price

db.commit()
db.refresh(db_item)

return {"message": f"已更新字段: {list(update_data.keys())}", "item": db_item}

数据库操作 – 删除

@app.delete("/items/{item_id}")
async def delete_item(item_id: int, db: Session = Depends(get_db)):
"""删除物品"""
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="物品不存在")

db.delete(db_item)
db.commit()

return {"message": f"物品 {item_id} 已删除"}

# —- 批量删除 —-
@app.delete("/items/batch")
async def delete_items_batch(item_ids: list[int], db: Session = Depends(get_db)):
"""批量删除物品"""
deleted_count = (
db.query(models.Item)
.filter(models.Item.id.in_(item_ids))
.delete(synchronize_session=False)
)
db.commit()

return {"message": f"成功删除 {deleted_count} 个物品"}

项目完整目录结构

fastapi_project/
├── main.py # 应用入口，定义路由
├── database.py # 数据库引擎和会话配置
├── models.py # SQLAlchemy ORM 模型
├── schemas.py # Pydantic 请求/响应模型
├── init_db.py # 数据库初始化脚本
├── requirements.txt # 依赖清单
└── templates/ # HTML 模板目录（可选）
└── index.html

requirements.txt：

fastapi>=0.110.0
uvicorn[standard]>=0.27.0
sqlalchemy>=2.0.0
pymysql>=1.1.0
pydantic>=2.0.0
jinja2>=3.1.0

总结

下一步推荐学习：异步数据库（async SQLAlchemy）、WebSocket、后台任务（BackgroundTasks）、OAuth2 + JWT 认证、Alembic 数据库迁移、Docker 部署。