FastAPI 路由请求

概述

路由是 Web 应用的核心组件，它定义了 URL 路径与处理函数之间的映射关系。FastAPI 提供了强大而灵活的路由系统，支持各种 HTTP 方法、路径参数、查询参数等。本章将详细介绍 FastAPI 的路由机制和请求处理。

🛣️ 基础路由概念

HTTP 方法支持

FastAPI 支持所有标准的 HTTP 方法：

python

from fastapi import FastAPI

app = FastAPI()

# GET 请求 - 获取资源
@app.get("/items")
async def get_items():
    return {"message": "获取所有物品"}

# POST 请求 - 创建资源
@app.post("/items")
async def create_item():
    return {"message": "创建新物品"}

# PUT 请求 - 更新资源（完整更新）
@app.put("/items/{item_id}")
async def update_item(item_id: int):
    return {"message": f"更新物品 {item_id}"}

# PATCH 请求 - 部分更新资源
@app.patch("/items/{item_id}")
async def patch_item(item_id: int):
    return {"message": f"部分更新物品 {item_id}"}

# DELETE 请求 - 删除资源
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": f"删除物品 {item_id}"}

# HEAD 请求 - 获取响应头（无响应体）
@app.head("/items/{item_id}")
async def head_item(item_id: int):
    return {"message": "HEAD 请求"}

# OPTIONS 请求 - 获取支持的方法
@app.options("/items")
async def options_items():
    return {"allowed_methods": ["GET", "POST", "PUT", "DELETE"]}

路由装饰器详解

python

from fastapi import FastAPI, status
from typing import List, Dict, Any

app = FastAPI()

# 基础路由
@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

# 带路径的路由
@app.get("/users")
async def get_users():
    return {"users": []}

# 嵌套路径
@app.get("/api/v1/users")
async def get_api_users():
    return {"api_version": "v1", "users": []}

# 指定响应状态码
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user():
    return {"message": "用户创建成功"}

# 指定响应模型和状态码
@app.get("/users/{user_id}", status_code=200)
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "John Doe"}

# 多个装饰器（同一函数处理多个路由）
@app.get("/health")
@app.get("/ping")
async def health_check():
    return {"status": "healthy"}

🎯 路径参数详解

基础路径参数

python

from fastapi import FastAPI, Path
from typing import Optional

app = FastAPI()

# 基础路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 多个路径参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: int):
    return {"user_id": user_id, "item_id": item_id}

# 字符串路径参数
@app.get("/items/{item_name}")
async def read_item_by_name(item_name: str):
    return {"item_name": item_name}

# 路径参数验证
@app.get("/items/{item_id}")
async def read_item_with_validation(
    item_id: int = Path(
        ...,  # 必需参数
        title="物品ID",
        description="要获取的物品的唯一标识符",
        ge=1,  # 大于等于 1
        le=1000  # 小于等于 1000
    )
):
    return {"item_id": item_id}

路径参数类型转换

python

from datetime import datetime
from uuid import UUID
from enum import Enum

# 枚举类型参数
class ItemType(str, Enum):
    ELECTRONICS = "electronics"
    CLOTHING = "clothing"
    BOOKS = "books"
    FOOD = "food"

@app.get("/items/{item_type}")
async def get_items_by_type(item_type: ItemType):
    return {
        "item_type": item_type,
        "message": f"获取 {item_type.value} 类型的物品"
    }

# UUID 参数
@app.get("/users/{user_id}")
async def get_user_by_uuid(user_id: UUID):
    return {"user_id": str(user_id)}

# 日期时间参数
@app.get("/orders/{order_date}")
async def get_orders_by_date(order_date: datetime):
    return {
        "order_date": order_date.isoformat(),
        "formatted_date": order_date.strftime("%Y-%m-%d %H:%M:%S")
    }

# 浮点数参数
@app.get("/prices/{price}")
async def get_price_info(price: float):
    return {
        "price": price,
        "tax": price * 0.1,
        "total": price * 1.1
    }

文件路径参数

python

# 捕获文件路径
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

# 示例：
# GET /files/documents/reports/2023/annual_report.pdf
# file_path = "documents/reports/2023/annual_report.pdf"

# 带验证的文件路径
@app.get("/static/{file_path:path}")
async def serve_static_file(
    file_path: str = Path(
        ...,
        title="文件路径",
        description="静态文件的相对路径",
        regex=r"^[a-zA-Z0-9._/\-]+$"  # 只允许特定字符
    )
):
    # 安全检查
    if ".." in file_path:
        raise HTTPException(status_code=400, detail="无效的文件路径")
    
    return {"file_path": file_path, "message": "文件访问成功"}

🔍 查询参数处理

基础查询参数

python

from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

# 基础查询参数
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 可选查询参数
@app.get("/items/")
async def read_items_with_query(
    skip: int = 0,
    limit: int = 10,
    q: Optional[str] = None
):
    items = {"skip": skip, "limit": limit}
    if q:
        items["q"] = q
    return items

# 查询参数验证
@app.get("/items/")
async def read_items_validated(
    skip: int = Query(0, ge=0, description="跳过的记录数"),
    limit: int = Query(10, ge=1, le=100, description="返回的记录数"),
    q: Optional[str] = Query(
        None,
        min_length=3,
        max_length=50,
        regex="^[a-zA-Z0-9\s]+$",
        description="搜索查询字符串"
    )
):
    return {"skip": skip, "limit": limit, "q": q}

高级查询参数

python

from typing import List, Set
from datetime import datetime, date

# 列表查询参数
@app.get("/items/")
async def read_items_with_list(
    tags: List[str] = Query([], description="标签列表"),
    categories: Set[str] = Query(set(), description="分类集合")
):
    return {
        "tags": tags,
        "categories": list(categories),
        "unique_categories": len(categories)
    }

# 调用示例：
# GET /items/?tags=electronics&tags=gadget&categories=tech&categories=gadget
# tags = ["electronics", "gadget"]
# categories = {"tech", "gadget"}

# 布尔查询参数
@app.get("/items/")
async def filter_items(
    is_available: bool = Query(True, description="是否可用"),
    is_featured: Optional[bool] = Query(None, description="是否特色商品"),
    on_sale: bool = False
):
    return {
        "is_available": is_available,
        "is_featured": is_featured,
        "on_sale": on_sale
    }

# 日期查询参数
@app.get("/orders/")
async def get_orders(
    start_date: Optional[date] = Query(None, description="开始日期"),
    end_date: Optional[date] = Query(None, description="结束日期"),
    created_after: Optional[datetime] = Query(None, description="创建时间之后")
):
    return {
        "start_date": start_date,
        "end_date": end_date,
        "created_after": created_after
    }

查询参数别名和弃用

python

# 参数别名
@app.get("/items/")
async def read_items_with_alias(
    item_query: Optional[str] = Query(
        None,
        alias="item-query",  # URL 中使用 item-query
        title="物品查询",
        description="搜索物品的查询字符串",
        deprecated=False
    )
):
    return {"item_query": item_query}

# 弃用参数
@app.get("/items/")
async def read_items_deprecated(
    q: Optional[str] = Query(None, description="搜索查询"),
    old_query: Optional[str] = Query(
        None,
        deprecated=True,  # 标记为已弃用
        description="旧的查询参数，请使用 q 参数"
    )
):
    # 兼容旧参数
    search_query = q or old_query
    return {"search_query": search_query}

🎭 路由组织和管理

使用 APIRouter

python

from fastapi import APIRouter, FastAPI

# 创建路由器
items_router = APIRouter(
    prefix="/items",
    tags=["items"],
    responses={404: {"description": "物品未找到"}}
)

users_router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "用户未找到"}}
)

# 物品相关路由
@items_router.get("/")
async def get_items():
    return {"items": []}

@items_router.get("/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

@items_router.post("/")
async def create_item():
    return {"message": "物品创建成功"}

# 用户相关路由
@users_router.get("/")
async def get_users():
    return {"users": []}

@users_router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# 主应用
app = FastAPI()

# 包含路由器
app.include_router(items_router, prefix="/api/v1")
app.include_router(users_router, prefix="/api/v1")

# 结果路由：
# GET /api/v1/items/
# GET /api/v1/items/{item_id}
# POST /api/v1/items/
# GET /api/v1/users/
# GET /api/v1/users/{user_id}

路由器嵌套

python

# 创建嵌套路由器
api_v1_router = APIRouter(prefix="/api/v1")
api_v2_router = APIRouter(prefix="/api/v2")

# v1 物品路由
v1_items_router = APIRouter(prefix="/items", tags=["v1-items"])

@v1_items_router.get("/")
async def get_items_v1():
    return {"version": "v1", "items": []}

# v2 物品路由
v2_items_router = APIRouter(prefix="/items", tags=["v2-items"])

@v2_items_router.get("/")
async def get_items_v2():
    return {"version": "v2", "items": [], "new_features": True}

# 组装路由
api_v1_router.include_router(v1_items_router)
api_v2_router.include_router(v2_items_router)

app = FastAPI()
app.include_router(api_v1_router)
app.include_router(api_v2_router)

# 结果路由：
# GET /api/v1/items/  -> v1 版本
# GET /api/v2/items/  -> v2 版本

🔧 路由配置和元数据

路由元数据

python

from fastapi import FastAPI, APIRouter
from typing import Dict, Any

app = FastAPI()

# 详细的路由配置
@app.get(
    "/items/{item_id}",
    summary="获取物品详情",
    description="根据物品ID获取详细信息，包括名称、价格、描述等",
    response_description="物品详情信息",
    tags=["物品管理"],
    operation_id="get_item_by_id",  # OpenAPI 操作ID
    deprecated=False,
    include_in_schema=True  # 是否包含在 OpenAPI 模式中
)
async def get_item(item_id: int):
    """
    获取物品详情：
    
    - **item_id**: 物品的唯一标识符
    
    返回物品的详细信息，包括：
    - 物品名称
    - 价格信息
    - 描述内容
    - 可用状态
    """
    return {"item_id": item_id, "name": "示例物品"}

# 路由器配置
admin_router = APIRouter(
    prefix="/admin",
    tags=["管理员"],
    dependencies=[],  # 依赖项列表
    responses={
        401: {"description": "未授权"},
        403: {"description": "权限不足"},
        500: {"description": "服务器错误"}
    }
)

@admin_router.get(
    "/stats",
    summary="获取统计信息",
    description="获取系统统计信息（需要管理员权限）"
)
async def get_admin_stats():
    return {"users": 100, "items": 500, "orders": 1200}

app.include_router(admin_router)

条件路由

python

import os
from fastapi import FastAPI

app = FastAPI()

# 根据环境添加路由
DEBUG = os.getenv("DEBUG", "false").lower() == "true"

@app.get("/")
async def read_root():
    return {"message": "生产环境"}

# 仅在调试模式下添加调试路由
if DEBUG:
    @app.get("/debug")
    async def debug_info():
        return {"debug": True, "environment": "development"}
    
    @app.get("/debug/config")
    async def debug_config():
        return {"config": dict(os.environ)}

# 功能开关路由
FEATURE_FLAGS = {
    "new_ui": os.getenv("ENABLE_NEW_UI", "false").lower() == "true",
    "beta_features": os.getenv("ENABLE_BETA", "false").lower() == "true"
}

if FEATURE_FLAGS["new_ui"]:
    @app.get("/ui/new")
    async def new_ui():
        return {"ui": "new_version"}

if FEATURE_FLAGS["beta_features"]:
    @app.get("/beta")
    async def beta_features():
        return {"features": ["feature_a", "feature_b"]}

🎪 高级路由特性

路由优先级和匹配

python

from fastapi import FastAPI

app = FastAPI()

# 路由顺序很重要！更具体的路由应该在前面

# 1. 最具体的路由
@app.get("/items/latest")
async def get_latest_items():
    return {"items": "latest"}

# 2. 特定值路由
@app.get("/items/count")
async def get_items_count():
    return {"count": 42}

# 3. 路径参数路由（放在最后）
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

# 错误示例（不要这样做）：
# 如果把 /items/{item_id} 放在前面，
# /items/latest 会被匹配为 item_id="latest"

路由冲突解决

python

from fastapi import FastAPI, HTTPException, Path
from typing import Union

app = FastAPI()

# 使用联合类型处理冲突
@app.get("/items/{item_identifier}")
async def get_item_flexible(item_identifier: str):
    # 尝试解析为整数ID
    if item_identifier.isdigit():
        item_id = int(item_identifier)
        return {"type": "id", "value": item_id}
    
    # 处理特殊字符串
    if item_identifier in ["latest", "popular", "featured"]:
        return {"type": "special", "value": item_identifier}
    
    # 处理常规字符串（可能是名称或代码）
    return {"type": "name", "value": item_identifier}

# 或者使用更严格的验证
@app.get("/items/by-id/{item_id}")
async def get_item_by_id(item_id: int = Path(..., ge=1)):
    return {"item_id": item_id}

@app.get("/items/by-name/{item_name}")
async def get_item_by_name(
    item_name: str = Path(..., regex=r"^[a-zA-Z0-9_-]+$")
):
    return {"item_name": item_name}

动态路由生成

python

from fastapi import FastAPI, APIRouter

def create_crud_router(resource_name: str, resource_data: dict):
    """动态创建 CRUD 路由"""
    router = APIRouter(prefix=f"/{resource_name}", tags=[resource_name])
    
    @router.get("/")
    async def list_resources():
        return {f"{resource_name}": list(resource_data.values())}
    
    @router.get("/{resource_id}")
    async def get_resource(resource_id: int):
        if resource_id in resource_data:
            return resource_data[resource_id]
        raise HTTPException(status_code=404, detail=f"{resource_name} not found")
    
    @router.post("/")
    async def create_resource(data: dict):
        new_id = max(resource_data.keys()) + 1 if resource_data else 1
        resource_data[new_id] = data
        return {f"{resource_name}_id": new_id, **data}
    
    return router

# 创建多个资源的路由
app = FastAPI()

# 用户资源
users_data = {1: {"name": "Alice"}, 2: {"name": "Bob"}}
users_router = create_crud_router("users", users_data)

# 产品资源
products_data = {1: {"name": "Laptop"}, 2: {"name": "Mouse"}}
products_router = create_crud_router("products", products_data)

# 注册路由
app.include_router(users_router, prefix="/api")
app.include_router(products_router, prefix="/api")

📊 路由性能优化

异步 vs 同步路由

python

import asyncio
import time
from fastapi import FastAPI

app = FastAPI()

# 同步路由（阻塞）
@app.get("/sync")
def sync_endpoint():
    # 模拟 CPU 密集型操作
    time.sleep(1)
    return {"message": "同步处理完成"}

# 异步路由（非阻塞）
@app.get("/async")
async def async_endpoint():
    # 模拟 I/O 操作
    await asyncio.sleep(1)
    return {"message": "异步处理完成"}

# 异步数据库操作示例
@app.get("/async-db")
async def async_db_operation():
    # 模拟异步数据库查询
    await asyncio.sleep(0.1)  # 模拟网络延迟
    return {"data": "来自异步数据库的数据"}

# 同步数据库操作示例（不推荐）
@app.get("/sync-db")
def sync_db_operation():
    # 模拟同步数据库查询
    time.sleep(0.1)  # 阻塞其他请求
    return {"data": "来自同步数据库的数据"}

路由缓存和优化

python

from functools import lru_cache
import asyncio

# 使用缓存优化计算密集型操作
@lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
    """计算斐波那契数列（缓存结果）"""
    if n <= 1:
        return n
    return expensive_computation(n-1) + expensive_computation(n-2)

@app.get("/fibonacci/{n}")
async def get_fibonacci(n: int):
    if n > 40:  # 防止过大的数值
        raise HTTPException(status_code=400, detail="数值过大")
    
    result = expensive_computation(n)
    return {"n": n, "fibonacci": result}

# 异步缓存示例
_cache = {}

async def async_expensive_operation(key: str):
    """模拟异步的昂贵操作"""
    if key in _cache:
        return _cache[key]
    
    # 模拟耗时操作
    await asyncio.sleep(0.5)
    result = f"结果: {key.upper()}"
    _cache[key] = result
    return result

@app.get("/cached/{key}")
async def get_cached_result(key: str):
    result = await async_expensive_operation(key)
    return {"key": key, "result": result, "cached": key in _cache}

总结

本章详细介绍了 FastAPI 的路由系统，包括：

✅ HTTP 方法支持：GET、POST、PUT、DELETE 等
✅ 路径参数：类型转换、验证、文件路径
✅ 查询参数：基础用法、验证、列表参数
✅ 路由组织：APIRouter、嵌套路由、模块化
✅ 路由配置：元数据、条件路由、动态生成
✅ 性能优化：异步处理、缓存策略

FastAPI 的路由系统非常灵活和强大，支持复杂的 URL 模式和参数验证。合理使用这些特性可以构建出既高性能又易维护的 API。

路由设计建议

遵循 RESTful 设计原则
使用清晰的 URL 结构
合理使用路径参数和查询参数
添加适当的验证和文档
考虑路由的执行顺序
优先使用异步处理 I/O 操作

在下一章中，我们将深入学习路径参数和查询参数的高级用法。

FastAPI 路由请求 ​

概述 ​

🛣️ 基础路由概念 ​

HTTP 方法支持 ​

路由装饰器详解 ​

🎯 路径参数详解 ​

基础路径参数 ​

路径参数类型转换 ​

文件路径参数 ​

🔍 查询参数处理 ​

基础查询参数 ​

高级查询参数 ​

查询参数别名和弃用 ​

🎭 路由组织和管理 ​

使用 APIRouter ​

路由器嵌套 ​

🔧 路由配置和元数据 ​

路由元数据 ​

条件路由 ​

🎪 高级路由特性 ​

路由优先级和匹配 ​

路由冲突解决 ​

动态路由生成 ​

📊 路由性能优化 ​

异步 vs 同步路由 ​

路由缓存和优化 ​

总结 ​

FastAPI 路由请求

概述

🛣️ 基础路由概念

HTTP 方法支持

路由装饰器详解

🎯 路径参数详解

基础路径参数

路径参数类型转换

文件路径参数

🔍 查询参数处理

基础查询参数

高级查询参数

查询参数别名和弃用

🎭 路由组织和管理

使用 APIRouter

路由器嵌套

🔧 路由配置和元数据

路由元数据

条件路由

🎪 高级路由特性

路由优先级和匹配

路由冲突解决

动态路由生成

📊 路由性能优化

异步 vs 同步路由

路由缓存和优化

总结