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

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

推荐阅读：FastAPI文档

---

目录

1. FastAPI 基础入门
   • FastAPI 框架基础
   • 参数分类（路径参数、查询参数、请求体参数）
   • 响应类型（JSON、HTML、文件、自定义格式）
   • 异常处理
2. FastAPI 进阶
   • 中间件与依赖注入
   • ORM 简介与使用
   • 数据库操作（查询、新增、更新、删除）

---

一、FastAPI 基础入门

1.1 FastAPI 框架基础

什么是 FastAPI？

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

核心特性

特性	说明
高性能	基于 Starlette（ASGI 框架）和 Pydantic，性能媲美 Node.js / Go
自动文档	自动生成 Swagger UI 和 ReDoc 交互式 API 文档
类型校验	基于 Python 类型提示，自动完成请求参数校验与序列化
依赖注入	内置强大的依赖注入系统，便于解耦和复用逻辑
异步支持	原生支持 async/await，适合高并发场景

环境搭建

# 安装 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 常用参数一览：

参数	类型	说明
...	Ellipsis	标记为必填
title	str	参数标题
description	str	参数描述
ge	int/float	大于等于（Greater than or Equal）
gt	int/float	大于（Greater Than）
le	int/float	小于等于（Less than or Equal）
lt	int/float	小于（Less Than）
min_length	int	字符串最小长度
max_length	int	字符串最大长度
regex	str	正则表达式匹配

正则表达式示例：

@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 值供快速测试

三种参数小结

参数类型	位置	适用场景	校验工具
路径参数	URL 路径	资源标识（ID、名称等）	Path
查询参数	URL ?key=value	过滤、排序、分页	Query
请求体参数	HTTP Body（JSON）	创建/更新复杂数据	Field（Pydantic）

---

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 类型参考：

文件类型	media_type
PDF	application/pdf
ZIP	application/zip
PNG	image/png
JPEG	image/jpeg
CSV	text/csv
Excel	application/vnd.openxmlformats-officedocument.spreadsheetml.sheet

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 → 客户端

中间件执行流程图：

注意：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": [...]}

依赖注入应用场景

场景	说明	示例
认证鉴权	统一校验 Token、权限	Depends(get_current_user)
数据库会话	自动管理数据库连接的创建和关闭	Depends(get_db)
分页参数	复用分页逻辑	Depends(pagination_params)
公共查询条件	提取通用过滤参数	Depends(common_filters)
配置注入	注入应用配置	Depends(get_settings)

实战：数据库会话依赖

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} 已删除"}

依赖注入执行链路图：

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

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

---

2.3 ORM 简介

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

ORM 分类

Python 生态中常见的 ORM 框架：

ORM 框架	特点	适用场景
SQLAlchemy	功能最强大，支持同步/异步，社区活跃	中大型项目（推荐）
Tortoise ORM	纯异步 ORM，API 类似 Django ORM	全异步项目
Peewee	轻量级，简单易用	小型项目
Django ORM	Django 内置，与 Django 深度集成	Django 项目

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

ORM 使用流程

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

ORM 请求处理完整流程图：

上图展示了一次完整的 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

---

总结

模块	核心要点
参数处理	路径参数（Path）、查询参数（Query）、请求体参数（Pydantic + Field）
响应类型	默认 JSON，支持 HTML / 文件 / 流式 / 自定义统一格式
异常处理	HTTPException + 自定义异常 + 全局异常处理器
中间件	CORS、日志、耗时统计等横切关注点
依赖注入	认证、数据库会话、分页等通用逻辑复用
ORM	SQLAlchemy 建表、CRUD 操作、条件查询、分页

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

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/J557565/article/details/161775385