FastAPI学习之路由

在 FastAPI 中，路由（Routing） 是将不同的 URL 路径映射到特定处理函数的核心机制。以下是 FastAPI 路由的详细解析，涵盖基本用法、路径参数、查询参数、请求方法等核心功能。

基本路由定义

通过装饰器 @app.get()、@app.post() 等绑定 HTTP 方法和路径：

from fastapi import FastAPI

app = FastAPI()

# GET 请求，路径为根目录 "/"
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

# POST 请求，路径为 "/items/"
@app.post("/items/")
async def create_item():
    return {"action": "item created"}

HTTP 方法

FastAPI 支持 所有标准 HTTP 方法，并通过简洁的装饰器语法实现路由绑定。以下是常用方法及其应用场景的详细说明：

支持的 HTTP 方法列表

核心方法详解与示例

GET – 获取资源

• 用途：查询数据库、读取文件、返回静态内容。
• 特点：幂等操作，不改变服务器状态。

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "name": "Sample Item"}

POST – 创建资源

• 用途：提交表单、上传文件、触发复杂操作。
• 特点：非幂等，每次请求可能产生新资源。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/")
async def create_item(item: Item):
    # 假设将 item 存入数据库
    return {"status": "created", "item": item}

PUT – 全量更新资源

• 用途：替换整个资源（需提供完整字段）。
• 特点：幂等操作，多次调用结果一致。

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    # 假设替换数据库中 item_id 对应的全部数据
    return {"item_id": item_id, "updated_item": item}

DELETE – 删除资源

• 用途：移除指定资源。
• 特点：幂等操作，删除后资源不再存在。

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    # 假设从数据库删除 item_id 对应的数据
    return {"status": "deleted", "item_id": item_id}

PATCH – 部分更新资源

• 用途：修改资源的特定字段（如用户昵称）。
• 特点：非幂等，需明确处理字段缺失逻辑。

from typing import Optional

class ItemUpdate(BaseModel):
    name: Optional[str] = None
    price: Optional[float] = None

@app.patch("/items/{item_id}")
async def partial_update(item_id: int, update: ItemUpdate):
    # 假设仅更新提供的字段
    return {"item_id": item_id, "updates": update}

其他方法使用场景

OPTIONS – 跨域预检请求

• 用途：配合 CORS 中间件处理跨域请求。
• 注意：通常由框架自动处理，无需手动实现。

@app.options("/items/")
async def get_options():
    return {"Allow": "GET, POST, OPTIONS"}

HEAD – 获取头信息

• 用途：检查资源是否存在或验证缓存。

@app.head("/items/{item_id}")
async def head_item(item_id: int):
    # 不返回响应体，仅检查资源是否存在
    return  # 无内容

高级用法

为同一路径绑定多个方法

@app.get("/tasks/{task_id}")
async def read_task(task_id: int):
    return {"method": "GET"}

@app.put("/tasks/{task_id}")
async def update_task(task_id: int):
    return {"method": "PUT"}

注意事项

• TRACE 方法：默认禁用，需显式启用（可能涉及安全风险）。
• 异步支持：所有方法均可使用async def 定义异步处理函数。
• RESTful 设计：建议遵循 REST 规范选择合适的方法（如 PUT vs PATCH）。

FastAPI 对 HTTP 方法的全面支持使其能够灵活构建 RESTful API 或自定义端点。通过结合 路径参数、查询参数 和 请求体模型，开发者可以高效实现各类业务逻辑。

路径参数（Path Parameters）

FastAPI 的 路径参数（Path Parameters） 允许从 URL 路径中直接提取动态值，是构建 RESTful API 的核心功能之一。

基本语法

在 URL 路径中用 {参数名} 定义动态参数，并在函数参数中声明对应的变量和类型：

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):  # 自动转换为整数
    return {"item_id": item_id}

• 访问/items/42 → 返回 {“item_id”: 42}
• 若传入非整数（如/items/apple），FastAPI 自动返回 422 验证错误。

类型验证与错误处理

FastAPI 支持多种类型，并自动验证数据格式：

支持的类型

• 基本类型：int,float, bool, str
• 复杂类型：UUID,datetime, Enum 等

from uuid import UUID

@app.get("/users/{user_uuid}")
async def get_user(user_uuid: UUID):
    return {"user_uuid": user_uuid}

• 有效 UUID 示例：550e8400-e29b-41d4-a716-446655440000

验证失败示例

• 请求/items/apple（item_id 期望整数）：

{
  "detail": [
    {
      "type": "int_parsing",
      "loc": ["path", "item_id"],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "apple"
    }
  ]
}

自定义错误消息

通过 Path 类的 description 参数或自定义异常来优化错误提示。

示例：添加描述信息

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(
        ...,
        ge=1,
        title="商品ID",
        description="商品ID必须是大于0的整数，且不超过1000",
        example=42
    )
):
    return {"item_id": item_id}

路径顺序优先级

路径顺序影响匹配逻辑，需将具体路径定义在通用路径之前：

@app.get("/users/me")
async def read_current_user():
    return {"user": "current user"}

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

• /users/me必须定义在 /users/{user_id} 之前，否则会被后者匹配。

路径参数校验

基础路径参数校验

通过类型提示（如 int、str）实现基本类型校验。

示例：类型校验

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):  # item_id 必须为整数
    return {"item_id": item_id}

效果：

• 访问/items/42 → 有效，返回 {“item_id”: 42}。
• 访问/items/apple → 无效，返回 422 错误（类型不匹配）。

使用 Path 类进行高级校验

通过 Path 类为路径参数添加额外校验规则（如最小值、最大值、正则表达式等）。

示例：数值范围校验

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="商品ID", ge=1, le=1000)  # ge=1 表示 ≥1，le=1000 表示 ≤1000
):
    return {"item_id": item_id}

参数说明：

• …（Ellipsis）：表示该参数是必填的。
• title：用于 API 文档的描述。
• ge（greater than or equal）：最小值。
• le（less than or equal）：最大值。
• gt（greater than）、lt（less than）：类似用法。

示例：字符串格式校验（正则表达式）

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_code}")
async def read_item(
    item_code: str = Path(..., regex=r"^[A-Z]{3}-\d{3}$")  # 格式必须为 "ABC-123"
):
    return {"item_code": item_code}

正则表达式规则：

• ^[A-Z]{3}-\d{3}$表示：3个大写字母 + 连字符 + 3个数字。

支持的验证参数：

正则表达式路径

通过 regex 参数限制路径格式：

@app.get("/files/{file_path:path}", tags=["files"])
async def read_file(
    file_path: str = Path(..., regex="^[a-zA-Z0-9_\-/]+$")
):
    return {"file_path": file_path}

• regex=”^[a-zA-Z0-9_\-/]+$”确保路径仅包含字母、数字、下划线、连字符和斜杠。

多路径参数与顺序控制

当路径中有多个参数时，需确保参数顺序正确。使用 * 强制关键字参数（Keyword-only）来避免歧义。

示例：多个路径参数

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/users/{user_id}/items/{item_id}")
async def read_item(
    *,  # 强制后面的参数必须通过关键字传递（避免顺序问题）
    user_id: int = Path(..., ge=1),
    item_id: int = Path(..., ge=1)
):
    return {"user_id": user_id, "item_id": item_id}

枚举路径参数

使用 Enum 类限制路径参数只能取预定义的值。

示例：枚举校验

from enum import Enum
from fastapi import FastAPI

app = FastAPI()

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 == ModelName.alexnet:
        return {"model": model_name, "message": "Deep Learning FTW!"}
    return {"model": model_name}

效果：

• 访问/models/alexnet → 有效。
• 访问/models/invalid → 无效，返回 422 错误。

复杂路径参数（如包含斜杠）

使用 :path 声明路径参数可以包含斜杠（/）。

示例：文件路径参数

from fastapi import FastAPI

app = FastAPI()

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

效果：

• 访问/files/home/user/file.txt → 返回 {“file_path”: “home/user/file.txt”}。

校验逻辑的执行顺序

FastAPI 按以下顺序校验路径参数：

• 类型转换：将字符串转换为声明的类型（如item_id: int）。
• Pydantic 校验：检查Path 类中定义的规则（如 ge=1）。
• 路由匹配：确保路径参数与路由模式匹配。

常见校验场景总结

查询参数（Query Parameters）

在 FastAPI 中，查询参数（Query Parameters） 是 URL 中位于 ? 后的键值对（如 /items?skip=10&limit=20），用于过滤、分页或配置请求。

基本语法

函数中未声明为路径参数的参数，默认被视为查询参数。

查询参数可以是 可选 或 必填，通过默认值区分：

可选参数（有默认值）

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

• 访问/items/ → 使用默认值：skip=0, limit=10
• 访问/items/?skip=20&limit=5 → skip=20, limit=5

必填参数（无默认值）

@app.get("/user/")
async def get_user(username: str):  # 无默认值 → 必填
    return {"username": username}

• 若未提供username（如 /user/），FastAPI 返回 422 错误：

{
  "detail": [
    {
      "type": "missing",
      "loc": ["query", "username"],
      "msg": "Field required"
    }
  ]
}

类型验证与约束

使用 Query 类为参数添加验证规则和文档描述：

字符串校验

from fastapi import Query

@app.get("/search/")
async def search(
    keyword: str = Query(
        min_length=2,          # 最小长度
        max_length=50,        # 最大长度
        regex="^[A-Za-z\s]+$" # 仅允许字母和空格
    )
):
    return {"keyword": keyword}

• 无效示例：/search/?keyword=a→ 错误：min_length=2
• 有效示例：/search/?keyword=fast api

数值范围验证

@app.get("/products/")
async def get_products(
    price_min: float = Query(ge=0),    # 值 >= 0
    price_max: float = Query(lt=1000)  # 值 < 1000
):
return {"price_min": price_min, "price_max": price_max}

正则表达式校验

@app.get("/validate/")
async def validate_code(
    code: str = Query(regex="^[A-Z]{3}-\d{4}$")  # 格式如 ABC-1234
):
    return {"code": code}

• 有效示例：/validate/?code=XYZ-9876
• 无效示例：/validate/?code=abc-123→ 错误：不匹配正则

高级用法

多值查询参数（列表）

允许通过重复键或逗号分隔传递多个值：

@app.get("/tags/")
async def get_items(tags: list[str] = Query()):
    return {"tags": tags}

• 访问/tags/?tags=python&tags=api → tags=[“python”, “api”]
• 或/tags/?tags=python,api → tags=[“python”, “api”]

参数别名 (Alias)

@app.get("/data/")
async def fetch_data(
    q: str = Query(..., alias="query-string")  # URL 中使用 query-string
):
    return {"query": q}

• 访问方式：/data/?query-string=test

混合路径参数和查询参数

@app.get("/users/{user_id}/orders")
async def get_orders(
    user_id: int,          # 路径参数
    status: str = "all",   # 查询参数（默认值 "all"）
    limit: int = 100
):
    return {"user_id": user_id, "status": status, "limit": limit}

• 访问/users/42/orders?status=paid&limit=5

强制必填参数

即使参数有默认值，也能标记为必填：

@app.get("/search/")
async def search(
    keyword: str = Query(..., min_length=2)  # ... 表示必填
):
    return {"results": f"Searching for '{keyword}'"}

枚举值约束

from enum import Enum

class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

@app.get("/items/")
async def sort_items(
    order: SortOrder = Query(..., description="排序方式")
):
    return {"order": order}

• 有效值：asc或 desc
• 无效值：/items/?order=invalid→ 错误：not a valid enumeration member

自定义错误消息

from fastapi import Query

@app.get("/custom-error/")
async def custom_validation(
    value: int = Query(gt=0, description="正整數", 
                      error_messages={
                          "gt": "数值必须大于0", 
                          "type": "需要整数类型"
                      })
):
    return {"value": value}

• 触发错误：/custom-error/?value=-5→ 返回自定义错误消息

常见问题与解决方案

参数名称冲突

• 问题：路径参数和查询参数同名（如/items/{id}?id=42）。
• 解决：使用alias 为查询参数设置别名：

@app.get("/items/{item_id}")
async def read_item(
    item_id: int,  # 路径参数
    id: str = Query(..., alias="item_query_id")  # 别名避免冲突
):
    return {"item_id": item_id, "query_id": id}

敏感数据暴露

• 问题：密码或 Token 通过 URL 传递（可能被浏览器历史记录）。
• 建议：敏感参数应使用 POST 请求体或 Headers 传递。

路由前缀（APIRouter）

FastAPI 的 APIRouter 是组织大型项目的核心工具，通过路由前缀和模块化管理，实现代码解耦和复用。

基础用法：创建路由前缀

定义路由组

在子模块（如 routers/users.py）中创建独立路由组：

from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["用户管理"])

@router.get("/", summary="获取用户列表")
async def get_users():
    return [{"id": 1, "name": "Alice"}]

@router.post("/", summary="创建用户")
async def create_user():
    return {"status": "created"}

主程序集成

在主文件（如 main.py）中挂载路由：

from fastapi import FastAPI
from routers import users  # 导入子模块

app = FastAPI()
app.include_router(users.router)

• 最终路由：
  • GET /users/ → 获取用户列表
  • POST /users/ → 创建用户
• Swagger 标签：自动归类到 “用户管理” 分组

核心配置参数

示例：带全局依赖的路由组

from fastapi import Depends, APIRouter
from .auth import get_current_user

router = APIRouter(
    prefix="/orders",
    tags=["订单"],
    dependencies=[Depends(get_current_user)],  # 全局身份验证
    responses={403: {"description": "Forbidden"}}
)

嵌套路由：多层级结构

通过多个 APIRouter 实现深度路径组织：

子模块 routers/admin/posts.py

from fastapi import APIRouter

router = APIRouter(prefix="/posts", tags=["管理员-文章"])

@router.get("/")
async def admin_get_posts():
    return {"message": "Admin posts"}

聚合路由 routers/admin.py

from fastapi import APIRouter
from .admin import posts, users

admin_router = APIRouter(prefix="/admin", tags=["管理员"])
admin_router.include_router(posts.router)
admin_router.include_router(users.router)

主程序挂载

from routers import admin

app.include_router(admin.admin_router)

• 最终路由：GET /admin/posts/→ 管理员文章列表

实用技巧

路由覆盖优先级

若主程序与子路由存在路径冲突，后挂载的路由会覆盖先前的：

app.include_router(public_router)  # /items/
app.include_router(admin_router)   # 若 admin_router 也有 /items/，则覆盖

动态前缀

通过环境变量动态配置前缀：

import os
from fastapi import APIRouter

API_PREFIX = os.getenv("API_PREFIX", "/api/v1")

router = APIRouter(prefix=API_PREFIX)

复用路径操作

在多个路由组中复用同一路径装饰器：

common_params = {"path": "/status", "tags": ["监控"]}

@router.get(**common_params)
async def get_status():
    return {"status": "OK"}

@router.post(**common_params)
async def update_status():
    return {"status": "Updated"}

最佳实践

模块化拆分

• 按业务功能拆分路由（如py, products.py, orders.py）
• 将路由组存放在routers/ 目录

统一响应模型

from pydantic import BaseModel

class ErrorResponse(BaseModel):
    code: int
    detail: str

router = APIRouter(
    responses={
        400: {"model": ErrorResponse},
        500: {"model": ErrorResponse}
    }
)

版本控制

v1_router = APIRouter(prefix="/api/v1")
v2_router = APIRouter(prefix="/api/v2")

# v1 路由
@v1_router.get("/users")
async def get_users_v1():
    return {"version": "v1"}

# v2 路由
@v2_router.get("/users")
async def get_users_v2():
    return {"version": "v2"}

常见问题

• 路径冲突
  • 现象：多个路由组定义了相同路径（如两个路由组都有/items/{id}）
  • 解决：调整include_router() 顺序或修改前缀
• 依赖不生效
  • 现象：子路由组的全局依赖未触发
  • 检查：确保主程序正确挂载路由，且依赖项已正确定义
• Swagger 标签混乱
  • 优化：为每个路由组明确指定tags，避免自动分类错误

通过合理使用 APIRouter，可以大幅提升项目的可维护性和扩展性，尤其适合中大型 FastAPI 应用架构。

路由依赖注入

在 FastAPI 中，路由依赖注入通过 Depends 实现，可以实现代码复用、请求预处理和权限控制。以下是完整的依赖注入使用指南，涵盖不同作用域的依赖配置和实际应用场景。

依赖注入基础

单个路由的依赖

from fastapi import Depends, FastAPI

app = FastAPI()

# 定义依赖函数
def get_db_session():
    print("创建数据库会话")
    db = "DBSession"
    yield db
    print("关闭数据库会话")

@app.get("/items/")
async def get_items(db: str = Depends(get_db_session)):
    return {"db": db}

访问效果：

• 请求/items/ 时自动创建和关闭数据库会话
• 返回结果包含数据库会话信息

类形式的依赖

class AuthChecker:
    def __init__(self, role: str):
        self.required_role = role
    
    def __call__(self, token: str = Header(...)):
        if token != self.required_role:
            raise HTTPException(403, "Forbidden")

admin_auth = AuthChecker("admin")

@app.get("/admin/")
async def admin_panel(_=Depends(admin_auth)):
    return {"message": "Admin Access"}

不同作用域的依赖

全局依赖（整个应用）

from fastapi import Depends, FastAPI

app = FastAPI(dependencies=[Depends(log_request_info)])

def log_request_info(request: Request):
    print(f"请求路径: {request.url}")

@app.get("/demo/")
async def demo():
    return {"message": "全局依赖示例"}

• 所有路由都会触发log_request_info

路由组依赖

from fastapi import APIRouter

router = APIRouter(
    dependencies=[Depends(verify_api_key)],  # 组内所有路由生效
    prefix="/v1"
)

def verify_api_key(api_key: str = Query(...)):
    if api_key != "secret123":
        raise HTTPException(403, "Invalid API Key")

@router.get("/data/")
async def get_data():
    return {"data": "敏感信息"}

• 访问/v1/data?api_key=secret123 才能获取数据

单个路由的依赖

from fastapi import Security
from .auth import get_current_user

@app.get("/user/profile/", dependencies=[Depends(get_current_user)])
async def user_profile():
    return {"profile": "私有数据"}

依赖执行顺序与覆盖

优先级顺序

依赖执行顺序为（从高到低）：

• 路由自身dependencies
• 路由组dependencies
• 全局dependencies

依赖覆盖示例

# 全局依赖
app = FastAPI(dependencies=[Depends(logging)])

# 路由组覆盖
admin_router = APIRouter(dependencies=[Depends(auth_admin)])

# 路由自身覆盖
@admin_router.get("/super/", dependencies=[Depends(super_auth)])
async def super_admin():
    return {"level": "最高权限"}

• 访问/super/ 时，执行顺序： logging → auth_admin → super_auth

实用场景示例

JWT 身份验证

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_jwt(token)  # 自定义解码逻辑
    if not user:
        raise HTTPException(401, "无效凭证")
    return user

@app.get("/protected/")
async def protected_route(user: dict = Depends(get_current_user)):
    return {"user": user["username"]}

数据库事务管理

async def get_db():
    async with AsyncSessionLocal() as session:
        async with session.begin():
            yield session

@app.post("/orders/")
async def create_order(
    order_data: OrderCreate,
    db: AsyncSession = Depends(get_db)
):
    order = Order(**order_data.dict())
    db.add(order)
    return {"id": order.id}

速率限制

from fastapi import HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

def rate_limit(limit: str = "10/minute"):
    def dependency(request: Request):
        if not limiter.check(limit, request):
            raise HTTPException(429, "请求过于频繁")
    return Depends(dependency)

@app.get("/api/", dependencies=[rate_limit("5/second")])
async def sensitive_api():
    return {"data": "高频接口数据"}

高级技巧

多依赖组合

def check_permission(permission: str):
    def checker(user: dict = Depends(get_current_user)):
        if permission not in user["permissions"]:
            raise HTTPException(403, "权限不足")
    return Depends(checker)

@app.delete("/resource/{id}")
async def delete_resource(
    _=Depends(check_permission("delete"))
):
    return {"status": "deleted"}

缓存依赖结果

from fastapi import Depends
from functools import lru_cache

@lru_cache
def get_config():
    return load_config()  # 耗时操作

@app.get("/settings/")
async def get_settings(config: dict = Depends(get_config)):
    return config

动态依赖生成

def dynamic_dep(env: str):
    if env == "prod":
        return Depends(prod_auth)
    else:
        return Depends(test_auth)

@app.get("/env/")
async def environment_route(_=Depends(dynamic_dep(os.getenv("ENV")))):
    return {"status": "access granted"}

常见问题解答

依赖项之间如何共享数据？

def get_user(request: Request):
    return User(**request.state.user_data)

def require_role(role: str):
    def checker(user: User = Depends(get_user)):
        if user.role != role:
            raise HTTPException(403, "角色不符")
    return Depends(checker)

如何跳过 Swagger 的依赖验证？

router = APIRouter(
    dependencies=[Depends(check_auth)],
    swagger_ui_parameters={"try_it_out_enabled": False}
)

依赖项抛出异常时的处理

FastAPI 会自动捕获依赖中的 HTTPException，生成对应的错误响应：

async def verify_token(token: str = Header(...)):
    if not validate(token):
        raise HTTPException(401, "Invalid token")

最佳实践总结

分层设计

• 全局依赖：日志记录、基础验证
• 路由组依赖：模块级权限控制
• 路由自身依赖：具体业务校验

依赖复用

# 在 common/dependencies.py 中集中管理
__all__ = ["get_db", "auth_required", "rate_limit"]

性能优化

• 对耗时依赖使用缓存 (@lru_cache)
• 避免在依赖中执行阻塞操作

安全建议

# 敏感操作使用 Security 替代 Depends
from fastapi.security import Security

def sensitive_dep():
    ...

@app.post("/transfer/")
async def money_transfer(_=Security(sensitive_dep, scopes=["finance"])):
    ...

通过合理设计依赖注入，可以实现清晰的代码结构、统一的权限管理和高效的系统资源利用。

路由高级用法

自定义响应状态码

from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item():
    return {"result": "created"}

路由文档定制

通过装饰器参数自定义 Swagger 文档：

@app.post(
    "/items/",
    summary="Create an item",  # 接口摘要
    description="Create a new item with name and price",  # 详细描述
    response_description="The created item"  # 响应说明
)
async def create_item(item: Item):
    return item