FastAPI 路由参数

概述

在FastAPI中，参数是构建灵活API的关键组件。本章将深入探讨路径参数、查询参数、请求体参数等各种参数类型，以及如何进行验证、转换和文档化。掌握这些知识将帮助您构建更加健壮和用户友好的API。

🎯 路径参数详解

基础路径参数

路径参数是URL路径中的变量部分，FastAPI自动进行类型转换和验证：

from fastapi import FastAPI, Path, HTTPException
from typing import Optional
from enum import Enum

app = FastAPI()

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

# 字符串路径参数
@app.get("/users/{username}")
async def read_user(username: str):
    return {"username": username}

# 浮点数路径参数
@app.get("/prices/{price}")
async def read_price(price: float):
    return {"price": price, "formatted": f"${price:.2f}"}

路径参数验证

使用Path类添加验证和文档：

from datetime import datetime
from uuid import UUID

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

# 字符串验证
@app.get("/users/{username}")
async def read_user_validated(
    username: str = Path(
        ...,
        title="用户名",
        description="用户的唯一用户名",
        min_length=3,
        max_length=20,
        regex="^[a-zA-Z0-9_]+$",
        example="john_doe"
    )
):
    return {"username": username}

# UUID参数
@app.get("/orders/{order_id}")
async def read_order(
    order_id: UUID = Path(
        ...,
        title="订单ID",
        description="订单的UUID标识符",
        example="123e4567-e89b-12d3-a456-426614174000"
    )
):
    return {"order_id": str(order_id)}

枚举路径参数

class ItemType(str, Enum):
    ELECTRONICS = "electronics"
    CLOTHING = "clothing"
    BOOKS = "books"
    FOOD = "food"

class Priority(int, Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    URGENT = 4

@app.get("/items/category/{category}")
async def get_items_by_category(
    category: ItemType = Path(
        ...,
        title="物品分类",
        description="物品的分类类型"
    )
):
    return {
        "category": category,
        "message": f"获取{category.value}分类的物品",
        "available_categories": [item.value for item in ItemType]
    }

@app.get("/tasks/{priority}")
async def get_tasks_by_priority(priority: Priority):
    priority_names = {
        Priority.LOW: "低优先级",
        Priority.MEDIUM: "中等优先级", 
        Priority.HIGH: "高优先级",
        Priority.URGENT: "紧急"
    }
    
    return {
        "priority": priority.value,
        "name": priority_names[priority],
        "tasks": f"获取{priority_names[priority]}任务"
    }

🔍 查询参数详解

基础查询参数

from typing import Optional, List, Set

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

# 必需查询参数
@app.get("/search/")
async def search_items(q: str):
    return {"query": q, "results": []}

# 布尔查询参数
@app.get("/items/")
async def filter_items(
    available: bool = True,
    featured: Optional[bool] = None
):
    return {
        "available": available,
        "featured": featured,
        "message": "过滤条件已应用"
    }

查询参数验证

from fastapi import Query

@app.get("/items/")
async def read_items_with_validation(
    skip: int = Query(
        0,
        title="跳过记录数",
        description="跳过的记录数量，用于分页",
        ge=0,
        example=0
    ),
    limit: int = Query(
        10,
        title="限制记录数",
        description="返回的最大记录数",
        ge=1,
        le=100,
        example=10
    ),
    q: Optional[str] = Query(
        None,
        title="搜索查询",
        description="搜索查询字符串",
        min_length=3,
        max_length=50,
        example="laptop"
    ),
    sort_by: str = Query(
        "created_at",
        title="排序字段",
        description="用于排序的字段名",
        regex="^(name|price|created_at|updated_at)$"
    )
):
    return {
        "skip": skip,
        "limit": limit,
        "q": q,
        "sort_by": sort_by
    }

列表查询参数

# 列表参数
@app.get("/items/")
async def read_items_with_lists(
    tags: List[str] = Query(
        [],
        title="标签列表",
        description="物品标签列表",
        example=["electronics", "mobile"]
    ),
    categories: Set[str] = Query(
        set(),
        title="分类集合",
        description="物品分类集合"
    ),
    prices: List[float] = Query(
        [],
        title="价格列表",
        description="价格范围列表"
    )
):
    return {
        "tags": tags,
        "categories": list(categories),
        "prices": prices,
        "filters_applied": len(tags) + len(categories) + len(prices)
    }

# 调用示例：
# GET /items/?tags=electronics&tags=mobile&categories=tech&prices=100.0&prices=200.0

查询参数别名和弃用

@app.get("/items/")
async def read_items_with_alias(
    # 使用别名
    item_query: Optional[str] = Query(
        None,
        alias="item-query",  # URL中使用连字符
        title="物品查询",
        description="搜索物品的查询字符串"
    ),
    # 弃用参数
    old_param: Optional[str] = Query(
        None,
        deprecated=True,
        title="旧参数",
        description="已弃用的参数，请使用item-query"
    ),
    # 隐藏参数（不在文档中显示）
    internal_param: Optional[str] = Query(
        None,
        include_in_schema=False,
        description="内部使用的参数"
    )
):
    # 兼容性处理
    search_query = item_query or old_param
    
    return {
        "search_query": search_query,
        "internal_param": internal_param
    }

📅 日期和时间参数

日期时间处理

from datetime import datetime, date, time
from typing import Optional

@app.get("/events/")
async def get_events(
    start_date: Optional[date] = Query(
        None,
        title="开始日期",
        description="事件开始日期 (YYYY-MM-DD)",
        example="2023-12-01"
    ),
    end_date: Optional[date] = Query(
        None,
        title="结束日期", 
        description="事件结束日期 (YYYY-MM-DD)",
        example="2023-12-31"
    ),
    created_after: Optional[datetime] = Query(
        None,
        title="创建时间之后",
        description="获取此时间之后创建的事件",
        example="2023-12-01T10:00:00"
    ),
    time_slot: Optional[time] = Query(
        None,
        title="时间段",
        description="事件时间段 (HH:MM:SS)",
        example="14:30:00"
    )
):
    # 日期验证
    if start_date and end_date and start_date > end_date:
        raise HTTPException(
            status_code=400,
            detail="开始日期不能晚于结束日期"
        )
    
    return {
        "start_date": start_date,
        "end_date": end_date,
        "created_after": created_after,
        "time_slot": time_slot,
        "date_range_days": (end_date - start_date).days if start_date and end_date else None
    }

🎭 混合参数类型

路径+查询+请求体参数

from pydantic import BaseModel

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

@app.put("/items/{item_id}")
async def update_item(
    # 路径参数
    item_id: int = Path(
        ...,
        title="物品ID",
        ge=1
    ),
    # 查询参数
    notify_users: bool = Query(
        False,
        title="通知用户",
        description="是否通知相关用户"
    ),
    # 请求体参数
    item: ItemUpdate = ...,
    # 可选查询参数
    reason: Optional[str] = Query(
        None,
        title="更新原因",
        max_length=200
    )
):
    update_data = item.dict(exclude_unset=True)
    
    result = {
        "item_id": item_id,
        "updated_fields": list(update_data.keys()),
        "notify_users": notify_users,
        "update_data": update_data
    }
    
    if reason:
        result["reason"] = reason
    
    return result

多个路径参数

@app.get("/users/{user_id}/orders/{order_id}/items/{item_id}")
async def get_user_order_item(
    user_id: int = Path(..., title="用户ID", ge=1),
    order_id: int = Path(..., title="订单ID", ge=1),
    item_id: int = Path(..., title="物品ID", ge=1),
    include_details: bool = Query(False, title="包含详情")
):
    return {
        "user_id": user_id,
        "order_id": order_id,
        "item_id": item_id,
        "include_details": include_details,
        "resource_path": f"/users/{user_id}/orders/{order_id}/items/{item_id}"
    }

🔧 参数验证和错误处理

自定义验证

from pydantic import validator, ValidationError

class SearchParams(BaseModel):
    query: str
    category: Optional[str] = None
    min_price: Optional[float] = None
    max_price: Optional[float] = None
    
    @validator('query')
    def query_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('查询字符串不能为空')
        return v.strip()
    
    @validator('max_price')
    def max_price_must_be_greater_than_min(cls, v, values):
        if 'min_price' in values and values['min_price'] is not None and v is not None:
            if v <= values['min_price']:
                raise ValueError('最大价格必须大于最小价格')
        return v

@app.get("/search/")
async def search_with_validation(
    query: str = Query(..., min_length=1),
    category: Optional[str] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0)
):
    try:
        # 使用Pydantic模型验证
        params = SearchParams(
            query=query,
            category=category,
            min_price=min_price,
            max_price=max_price
        )
        
        return {
            "search_params": params.dict(),
            "message": "搜索参数验证通过"
        }
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())

条件参数验证

@app.get("/reports/")
async def generate_report(
    report_type: str = Query(
        ...,
        regex="^(daily|weekly|monthly|yearly)$",
        title="报告类型"
    ),
    start_date: Optional[date] = Query(None, title="开始日期"),
    end_date: Optional[date] = Query(None, title="结束日期"),
    format: str = Query(
        "json",
        regex="^(json|csv|pdf)$",
        title="输出格式"
    )
):
    # 根据报告类型验证日期要求
    if report_type in ["daily", "weekly"] and not start_date:
        raise HTTPException(
            status_code=400,
            detail=f"{report_type}报告需要提供开始日期"
        )
    
    if start_date and end_date:
        if start_date > end_date:
            raise HTTPException(
                status_code=400,
                detail="开始日期不能晚于结束日期"
            )
        
        # 检查日期范围
        date_diff = (end_date - start_date).days
        max_days = {"daily": 31, "weekly": 90, "monthly": 365, "yearly": 1095}
        
        if date_diff > max_days.get(report_type, 365):
            raise HTTPException(
                status_code=400,
                detail=f"{report_type}报告的日期范围不能超过{max_days[report_type]}天"
            )
    
    return {
        "report_type": report_type,
        "start_date": start_date,
        "end_date": end_date,
        "format": format,
        "estimated_records": date_diff if start_date and end_date else "未知"
    }

📊 参数转换和处理

自定义参数转换器

def parse_coordinates(coord_str: str) -> tuple:
    """解析坐标字符串 '123.456,789.012' 为元组"""
    try:
        lat, lng = map(float, coord_str.split(','))
        if not (-90 <= lat <= 90):
            raise ValueError("纬度必须在-90到90之间")
        if not (-180 <= lng <= 180):
            raise ValueError("经度必须在-180到180之间")
        return (lat, lng)
    except (ValueError, TypeError) as e:
        raise HTTPException(status_code=400, detail=f"坐标格式错误: {str(e)}")

@app.get("/locations/nearby/")
async def find_nearby_locations(
    coordinates: str = Query(
        ...,
        title="坐标",
        description="格式: '纬度,经度' 例如: '39.9042,116.4074'",
        example="39.9042,116.4074"
    ),
    radius: float = Query(
        1.0,
        title="搜索半径",
        description="搜索半径（公里）",
        ge=0.1,
        le=100.0
    )
):
    lat, lng = parse_coordinates(coordinates)
    
    return {
        "center": {"latitude": lat, "longitude": lng},
        "radius_km": radius,
        "search_area": f"以({lat}, {lng})为中心，半径{radius}公里的区域"
    }

复杂查询参数处理

from urllib.parse import unquote

@app.get("/advanced-search/")
async def advanced_search(
    # JSON字符串参数
    filters: Optional[str] = Query(
        None,
        title="过滤条件",
        description="JSON格式的过滤条件",
        example='{"category": "electronics", "brand": "apple"}'
    ),
    # 排序参数
    sort: str = Query(
        "created_at:desc",
        title="排序",
        description="排序字段和方向，格式: field:direction",
        example="price:asc"
    ),
    # 分页参数
    page: int = Query(1, title="页码", ge=1),
    per_page: int = Query(20, title="每页数量", ge=1, le=100)
):
    import json
    
    # 解析过滤条件
    parsed_filters = {}
    if filters:
        try:
            parsed_filters = json.loads(unquote(filters))
        except json.JSONDecodeError:
            raise HTTPException(
                status_code=400,
                detail="过滤条件JSON格式错误"
            )
    
    # 解析排序
    try:
        sort_field, sort_direction = sort.split(':')
        if sort_direction not in ['asc', 'desc']:
            raise ValueError("排序方向必须是asc或desc")
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail="排序格式错误，应为 'field:direction'"
        )
    
    # 计算偏移量
    offset = (page - 1) * per_page
    
    return {
        "filters": parsed_filters,
        "sort": {"field": sort_field, "direction": sort_direction},
        "pagination": {
            "page": page,
            "per_page": per_page,
            "offset": offset
        },
        "query_summary": f"第{page}页，每页{per_page}条，按{sort_field}排序"
    }

🎯 最佳实践和技巧

参数组织和复用

from fastapi import Depends

# 可复用的参数依赖
class PaginationParams:
    def __init__(
        self,
        page: int = Query(1, ge=1, title="页码"),
        per_page: int = Query(20, ge=1, le=100, title="每页数量")
    ):
        self.page = page
        self.per_page = per_page
        self.offset = (page - 1) * per_page

class SortParams:
    def __init__(
        self,
        sort_by: str = Query("created_at", title="排序字段"),
        sort_order: str = Query("desc", regex="^(asc|desc)$", title="排序方向")
    ):
        self.sort_by = sort_by
        self.sort_order = sort_order

# 使用参数依赖
@app.get("/items/")
async def list_items(
    pagination: PaginationParams = Depends(),
    sorting: SortParams = Depends(),
    search: Optional[str] = Query(None, title="搜索关键词")
):
    return {
        "pagination": {
            "page": pagination.page,
            "per_page": pagination.per_page,
            "offset": pagination.offset
        },
        "sorting": {
            "sort_by": sorting.sort_by,
            "sort_order": sorting.sort_order
        },
        "search": search
    }

参数文档化

@app.get(
    "/products/{product_id}",
    summary="获取产品详情",
    description="根据产品ID获取详细的产品信息",
    response_description="产品详情信息"
)
async def get_product(
    product_id: int = Path(
        ...,
        title="产品ID",
        description="要获取的产品的唯一标识符",
        example=123,
        ge=1
    ),
    include_reviews: bool = Query(
        False,
        title="包含评论",
        description="是否在响应中包含产品评论"
    ),
    review_limit: Optional[int] = Query(
        5,
        title="评论数量限制",
        description="返回的评论数量限制（仅在include_reviews=true时有效）",
        ge=1,
        le=50
    )
):
    """
    获取产品详情
    
    此端点返回指定产品的详细信息，包括：
    - 产品基本信息（名称、价格、描述）
    - 产品规格和属性
    - 可选的用户评论（如果启用）
    
    **参数说明：**
    - **product_id**: 产品的数据库ID
    - **include_reviews**: 设置为true时包含用户评论
    - **review_limit**: 限制返回的评论数量
    
    **示例请求：**
    ```
    GET /products/123?include_reviews=true&review_limit=10
    ```
    """
    product_data = {
        "product_id": product_id,
        "name": "示例产品",
        "price": 99.99,
        "description": "这是一个示例产品"
    }
    
    if include_reviews:
        product_data["reviews"] = [
            {"id": i, "rating": 5, "comment": f"评论 {i}"}
            for i in range(1, min(review_limit + 1, 6))
        ]
    
    return product_data

总结

本章详细介绍了FastAPI中各种参数类型的使用：

• ✅ 路径参数：基础用法、验证、枚举类型
• ✅ 查询参数：验证、列表参数、别名和弃用
• ✅ 日期时间参数：日期、时间、日期时间处理
• ✅ 混合参数：路径+查询+请求体的组合使用
• ✅ 参数验证：自定义验证、条件验证、错误处理
• ✅ 参数转换：自定义转换器、复杂参数处理
• ✅ 最佳实践：参数组织、复用、文档化

掌握这些参数处理技巧将帮助您构建更加灵活、健壮和用户友好的API接口。

参数设计建议

• 使用清晰的参数名称和描述
• 添加适当的验证和约束
• 提供有意义的示例值
• 考虑参数的向后兼容性
• 合理使用默认值
• 编写详细的API文档

在下一章中，我们将学习FastAPI的请求和响应处理，包括请求体、响应模型和状态码管理。