FastAPI路径参数与查询参数详解

📂 所属阶段:第一阶段 — 快速筑基(基础篇)
🔗 相关章节:FastAPI简介与优势 · 环境搭建 · Pydantic综合指南

目录

参数处理概述

在Web开发中,参数处理是构建API的核心环节。FastAPI提供了强大而灵活的参数处理机制,基于Python类型提示实现自动验证和转换。

FastAPI参数类型

FastAPI支持多种参数类型:

  1. 路径参数 (Path Parameters) - 从URL路径中提取
  2. 查询参数 (Query Parameters) - 从URL查询字符串中提取
  3. 请求体参数 (Request Body) - 从请求体中提取
  4. Cookie参数 (Cookie Parameters) - 从Cookie中提取
  5. Header参数 (Header Parameters) - 从请求头中提取

基础示例

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 路径参数示例
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

# 查询参数示例
@app.get("/items/")
def get_items(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

路径参数详解

基础路径参数

路径参数是从URL路径中提取的参数,使用花括号 {} 定义:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    """
    根据ID获取项目
    路径参数会自动进行类型转换和验证
    """
    return {"item_id": item_id}

路径参数类型

FastAPI支持多种类型的路径参数:

from fastapi import FastAPI
from enum import Enum
from uuid import UUID

app = FastAPI()

# 整数类型
@app.get("/items/{item_id}")
def get_item_by_id(item_id: int):
    return {"item_id": item_id}

# 字符串类型
@app.get("/users/{user_name}")
def get_user_by_name(user_name: str):
    return {"user_name": user_name}

# UUID类型
@app.get("/orders/{order_uuid}")
def get_order_by_uuid(order_uuid: UUID):
    return {"order_uuid": order_uuid}

# 枚举类型
class ItemType(str, Enum):
    electronics = "electronics"
    books = "books"
    clothing = "clothing"

@app.get("/categories/{item_type}")
def get_category(item_type: ItemType):
    return {"item_type": item_type}

路径参数验证

使用Pydantic的Path进行高级验证:

from fastapi import FastAPI, Path
from typing import Annotated

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[
        int, 
        Path(
            title="项目ID", 
            description="项目的唯一标识符", 
            ge=1,  # greater than or equal to 1
            le=1000,  # less than or equal to 1000
            example=123
        )
    ]
):
    """
    获取项目信息
    - 参数验证:ID必须在1-1000范围内
    - 自动类型转换:字符串转整数
    """
    return {"item_id": item_id}

@app.get("/users/{user_id}/items/{item_id}")
def read_user_item(
    user_id: Annotated[int, Path(title="用户ID", ge=1)],
    item_id: Annotated[int, Path(title="项目ID", ge=1)],
    q: str = None
):
    """
    获取用户特定项目的详细信息
    支持多个路径参数
    """
    return {
        "user_id": user_id,
        "item_id": item_id,
        "q": q
    }

路径参数约束

from fastapi import FastAPI, Path
from typing import Annotated
import re

app = FastAPI()

# 正则表达式约束
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    """
    读取文件路径
    :path 表示匹配整个路径,包括斜杠
    """
    return {"file_path": file_path}

# 复杂路径参数验证
@app.get("/products/{product_code}")
def get_product(
    product_code: Annotated[
        str,
        Path(
            title="产品代码",
            regex=r"^[A-Z]{2,4}-\d{3,5}$",  # 格式:XX-XXX 或 XXXX-XXXXX
            min_length=5,
            max_length=10
        )
    ]
):
    """
    根据产品代码获取产品信息
    产品代码格式:字母-数字,如 AB-123 或 ABCD-12345
    """
    return {"product_code": product_code}

查询参数详解

基础查询参数

查询参数是从URL查询字符串中提取的参数:

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/items/")
def read_items(q: Optional[str] = None):
    """
    获取项目列表
    q: 可选的搜索查询参数
    """
    if q:
        return {"q": q, "message": f"Searching for {q}"}
    return {"message": "No search query"}

# 多个查询参数
@app.get("/users/")
def read_users(
    skip: int = 0,
    limit: int = 100,
    active: bool = True,
    sort_by: str = "id"
):
    """
    获取用户列表
    支持分页、筛选和排序
    """
    return {
        "skip": skip,
        "limit": limit,
        "active": active,
        "sort_by": sort_by
    }

查询参数验证

使用Pydantic的Query进行高级验证:

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

app = FastAPI()

@app.get("/items/")
def read_items(
    q: Annotated[
        Optional[str],
        Query(
            title="搜索查询",
            description="搜索关键词,最少3个字符,最多50个字符",
            min_length=3,
            max_length=50,
            regex=r"^[a-zA-Z0-9\s]+$"  # 只允许字母、数字和空格
        )
    ] = None,
    category: Annotated[
        Optional[str],
        Query(
            title="分类",
            description="产品分类,支持多选",
            alias="cat"  # URL参数别名
        )
    ] = None,
    tags: Annotated[
        List[str],
        Query(
            title="标签",
            description="产品标签,支持多个",
            min_items=1,
            max_items=5
        )
    ] = []
):
    """
    高级搜索功能
    支持搜索、分类筛选和标签筛选
    """
    return {
        "q": q,
        "category": category,
        "tags": tags
    }

查询参数约束

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

app = FastAPI()

# 数值范围约束
@app.get("/products/")
def filter_products(
    min_price: Annotated[
        Optional[float],
        Query(
            title="最低价格",
            ge=0,  # greater than or equal to 0
            description="价格下限,必须大于等于0"
        )
    ] = None,
    max_price: Annotated[
        Optional[float],
        Query(
            title="最高价格",
            gt=0,  # greater than 0
            description="价格上限,必须大于0"
        )
    ] = None,
    rating: Annotated[
        Optional[float],
        Query(
            title="评分",
            ge=0,
            le=5,
            description="产品评分,0-5分"
        )
    ] = None
):
    """
    价格和评分筛选
    支持价格区间和评分筛选
    """
    filters = {}
    if min_price is not None:
        filters["min_price"] = min_price
    if max_price is not None:
        filters["max_price"] = max_price
    if rating is not None:
        filters["rating"] = rating
    
    return {"filters": filters, "message": "Filter applied"}

# 分页参数
@app.get("/data/")
def get_paginated_data(
    page: Annotated[
        int,
        Query(
            ge=1,
            description="页码,从1开始"
        )
    ] = 1,
    page_size: Annotated[
        int,
        Query(
            ge=1,
            le=100,
            description="每页数量,1-100"
        )
    ] = 10,
    sort: Annotated[
        Optional[str],
        Query(
            regex=r"^[a-zA-Z_]+:(asc|desc)$",  # 格式:field:(asc|desc)
            description="排序方式,格式:字段名:(asc|desc)"
        )
    ] = None
):
    """
    分页和排序
    支持自定义分页和排序
    """
    offset = (page - 1) * page_size
    return {
        "page": page,
        "page_size": page_size,
        "offset": offset,
        "sort": sort
    }

请求体参数

基础请求体

使用Pydantic模型定义请求体结构:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

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

@app.post("/items/")
def create_item(item: Item):
    """
    创建新项目
    请求体自动验证和序列化
    """
    return item

# 带响应模型的请求
@app.post("/items-with-total/")
def create_item_with_total(item: Item):
    """
    创建项目并计算总价
    """
    item_dict = item.model_dump()
    if item.tax:
        total = item.price + item.tax
        item_dict["total"] = total
    return item_dict

复杂请求体结构

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

app = FastAPI()

class Address(BaseModel):
    street: str = Field(..., min_length=5, max_length=100)
    city: str
    country: str
    postal_code: str = Field(regex=r"^\d{5}(-\d{4})?$")  # ZIP码格式

class User(BaseModel):
    id: Optional[int] = None
    name: str = Field(..., min_length=2, max_length=50)
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    age: Optional[int] = Field(None, ge=0, le=150)
    addresses: List[Address] = []
    created_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = True

@app.post("/users/")
def create_user(user: User):
    """
    创建用户,支持多地址
    复杂嵌套模型验证
    """
    return user

多个请求参数

from fastapi import FastAPI, Body
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None

class User(BaseModel):
    username: str
    email: str

# 路径参数 + 查询参数 + 请求体
@app.put("/users/{user_id}/items/{item_id}")
def update_user_item(
    user_id: int,
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(..., ge=0, le=5, description="重要性等级 0-5"),
    reason: Optional[str] = Body(None, embed=True, description="更新原因")
):
    """
    更新用户特定项目
    组合多种参数类型
    """
    return {
        "user_id": user_id,
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance,
        "reason": reason
    }

参数验证与约束

内置验证器

from fastapi import FastAPI, Query, Path
from pydantic import BaseModel, Field, validator
from typing import Annotated, Optional
import re

app = FastAPI()

# 使用Field进行字段验证
class Product(BaseModel):
    name: str = Field(
        ..., 
        min_length=3, 
        max_length=100, 
        description="产品名称,3-100字符"
    )
    price: float = Field(
        ..., 
        gt=0, 
        description="产品价格,必须大于0"
    )
    category: str = Field(
        ..., 
        regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$",  # 有效的标识符格式
        description="产品分类,字母数字下划线"
    )
    tags: list = Field(
        default=[], 
        min_items=0, 
        max_items=10,
        description="标签列表,最多10个"
    )
    rating: float = Field(
        default=0.0, 
        ge=0, 
        le=5,
        description="评分,0-5分"
    )

@app.post("/products/")
def create_product(product: Product):
    """
    创建产品,应用多种验证规则
    """
    return product

# 查询参数验证
@app.get("/search/")
def advanced_search(
    q: Annotated[
        str,
        Query(
            min_length=2,
            max_length=100,
            description="搜索关键词,2-100字符"
        )
    ],
    price_min: Annotated[
        Optional[float],
        Query(
            ge=0,
            description="最低价格"
        )
    ] = None,
    price_max: Annotated[
        Optional[float],
        Query(
            gt=0,
            description="最高价格"
        )
    ] = None,
    category: Annotated[
        Optional[str],
        Query(
            regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$",
            description="分类名称格式验证"
        )
    ] = None
):
    """
    高级搜索功能
    支持多重验证和约束
    """
    criteria = {"q": q}
    if price_min is not None:
        criteria["price_min"] = price_min
    if price_max is not None:
        criteria["price_max"] = price_max
    if category is not None:
        criteria["category"] = category
    
    return {"criteria": criteria, "message": "Search executed"}

自定义验证器

from fastapi import FastAPI
from pydantic import BaseModel, Field, validator, root_validator
from typing import Optional
from datetime import datetime, date

app = FastAPI()

class Booking(BaseModel):
    customer_name: str
    check_in: date
    check_out: date
    room_type: str
    guests: int = Field(ge=1, le=10)
    
    @validator('customer_name')
    def validate_customer_name(cls, v):
        """验证客户姓名"""
        if not v or len(v.strip()) < 2:
            raise ValueError('客户姓名至少需要2个字符')
        if not v.replace(' ', '').isalpha():
            raise ValueError('客户姓名只能包含字母和空格')
        return v.strip().title()
    
    @validator('room_type')
    def validate_room_type(cls, v):
        """验证房间类型"""
        allowed_types = ['single', 'double', 'suite', 'deluxe']
        if v.lower() not in allowed_types:
            raise ValueError(f'房间类型必须是 {allowed_types} 之一')
        return v.lower()
    
    @root_validator
    def validate_dates(cls, values):
        """验证日期逻辑"""
        check_in = values.get('check_in')
        check_out = values.get('check_out')
        
        if check_in and check_out:
            if check_in >= check_out:
                raise ValueError('退房日期必须晚于入住日期')
            
            # 不能预订过去的时间
            if check_in < date.today():
                raise ValueError('不能预订过去的日期')
        
        return values

@app.post("/bookings/")
def create_booking(booking: Booking):
    """
    创建预订,应用自定义验证逻辑
    """
    # 计算住宿天数
    nights = (booking.check_out - booking.check_in).days
    booking_dict = booking.model_dump()
    booking_dict['nights'] = nights
    
    return booking_dict

Pydantic模型验证

基础模型验证

from fastapi import FastAPI
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, List
import re

app = FastAPI()

class EmailModel(BaseModel):
    """邮箱验证模型"""
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    
    # 自定义验证器
    @validator('email')
    def validate_email_format(cls, v):
        return v.lower()

class UserProfile(BaseModel):
    """用户资料模型"""
    username: str = Field(
        ..., 
        min_length=3, 
        max_length=50,
        regex=r"^[a-zA-Z0-9_]+$",  # 只允许字母数字下划线
        description="用户名,3-50字符,字母数字下划线"
    )
    email: str = Field(..., description="有效的电子邮箱")
    age: Optional[int] = Field(None, ge=0, le=150, description="年龄,0-150")
    bio: Optional[str] = Field(None, max_length=500, description="个人简介,最多500字符")
    interests: List[str] = Field(
        default=[], 
        min_items=0, 
        max_items=10,
        description="兴趣爱好,最多10个"
    )
    
    class Config:
        # 配置选项
        anystr_strip_whitespace = True  # 自动去除字符串首尾空白
        validate_assignment = True      # 赋值时验证
        str_strip_whitespace = True     # 自动去除字符串空白

@app.post("/profiles/")
def create_profile(profile: UserProfile):
    """
    创建用户资料
    应用全面的验证规则
    """
    return profile

高级模型特性

from fastapi import FastAPI
from pydantic import BaseModel, Field, validator, root_validator, PrivateAttr
from typing import Optional, Dict, Any
from datetime import datetime

app = FastAPI()

class AdvancedUser(BaseModel):
    """高级用户模型"""
    id: Optional[int] = Field(None, description="用户ID")
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    full_name: Optional[str] = Field(None, max_length=100)
    
    # 字段别名
    phone_number: str = Field(..., alias="phone", description="电话号码")
    
    # 默认工厂函数
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    
    # 额外字段处理
    class Config:
        allow_population_by_field_name = True  # 允许使用字段名或别名
        use_enum_values = True                # 使用枚举值
        validate_all = True                   # 验证所有字段
        extra = "forbid"                      # 禁止额外字段
    
    # 私有属性
    _internal_id: str = PrivateAttr(default=None)
    
    def __init__(self, **data):
        super().__init__(**data)
        self._internal_id = f"usr_{self.id or hash(self.username)}"

@app.post("/advanced-users/")
def create_advanced_user(user: AdvancedUser):
    """
    创建高级用户
    展示Pydantic高级特性
    """
    user_dict = user.model_dump()
    user_dict['_internal_id'] = user._internal_id
    return user_dict

参数组合使用

混合参数类型

from fastapi import FastAPI, Query, Path, Cookie, Header
from pydantic import BaseModel
from typing import Annotated, Optional

app = FastAPI()

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

@app.put("/users/{user_id}/items/{item_id}")
def update_user_item(
    user_id: Annotated[
        int,
        Path(title="用户ID", ge=1, description="目标用户ID")
    ],
    item_id: Annotated[
        int,
        Path(title="项目ID", ge=1, description="目标项目ID")
    ],
    item_update: ItemUpdate,
    priority: Annotated[
        int,
        Query(ge=1, le=5, description="更新优先级 1-5")
    ] = 3,
    x_token: Annotated[
        Optional[str],
        Header(description="认证令牌")
    ] = None,
    session_id: Annotated[
        Optional[str],
        Cookie(description="会话ID")
    ] = None,
    include_details: Annotated[
        bool,
        Query(description="是否包含详细信息")
    ] = False
):
    """
    综合参数使用示例
    演示路径参数、查询参数、请求体、Header、Cookie的组合使用
    """
    result = {
        "user_id": user_id,
        "item_id": item_id,
        "item_update": item_update,
        "priority": priority,
        "include_details": include_details
    }
    
    if x_token:
        result["authenticated"] = True
    if session_id:
        result["session_id"] = session_id
    
    return result

参数依赖和预处理

from fastapi import FastAPI, Depends, Query
from typing import Annotated

app = FastAPI()

# 参数预处理依赖
def validate_and_preprocess_query(
    q: Annotated[
        str,
        Query(min_length=1, max_length=100, description="搜索查询")
    ]
) -> str:
    """预处理搜索查询"""
    # 去除多余空格并转小写
    processed_q = ' '.join(q.split()).lower()
    return processed_q

def get_pagination_params(
    skip: Annotated[
        int,
        Query(ge=0, description="跳过的项目数")
    ] = 0,
    limit: Annotated[
        int,
        Query(ge=1, le=100, description="返回的项目数")
    ] = 10
) -> dict:
    """获取分页参数"""
    return {"skip": skip, "limit": limit}

@app.get("/search/")
def search_with_dependencies(
    processed_query: str = Depends(validate_and_preprocess_query),
    pagination: dict = Depends(get_pagination_params)
):
    """
    使用依赖注入的搜索功能
    演示参数预处理和依赖注入
    """
    return {
        "query": processed_query,
        "pagination": pagination,
        "message": f"Searching for '{processed_query}'"
    }

高级参数处理

动态参数处理

from fastapi import FastAPI, Query
from typing import Annotated, Dict, Any, Optional
from pydantic import BaseModel

app = FastAPI()

class DynamicFilter(BaseModel):
    """动态过滤器模型"""
    field: str
    operator: str
    value: str

@app.post("/dynamic-search/")
def dynamic_search(
    filters: Annotated[
        list,
        Query(description="动态过滤条件列表")
    ],
    sort_by: Annotated[
        Optional[str],
        Query(description="排序字段")
    ] = None,
    ascending: Annotated[
        bool,
        Query(description="升序排序")
    ] = True
):
    """
    动态搜索功能
    支持动态过滤条件
    """
    # 解析过滤条件
    parsed_filters = []
    for i in range(0, len(filters), 3):  # 假设每3个参数组成一个过滤条件
        if i + 2 < len(filters):
            parsed_filters.append({
                "field": filters[i],
                "operator": filters[i + 1],
                "value": filters[i + 2]
            })
    
    return {
        "filters": parsed_filters,
        "sort_by": sort_by,
        "ascending": ascending,
        "total_results": len(parsed_filters)  # 示例计数
    }

文件上传参数

from fastapi import FastAPI, File, UploadFile, Form
from typing import Annotated, Optional

app = FastAPI()

@app.post("/upload/")
async def upload_file_with_metadata(
    file: Annotated[
        UploadFile,
        File(description="要上传的文件")
    ],
    filename: Annotated[
        str,
        Form(description="文件名")
    ],
    description: Annotated[
        Optional[str],
        Form(description="文件描述")
    ] = None,
    category: Annotated[
        str,
        Form(description="文件分类")
    ] = "general"
):
    """
    文件上传功能
    结合文件和表单参数
    """
    # 读取文件内容
    content = await file.read()
    file_size = len(content)
    
    return {
        "filename": filename,
        "original_filename": file.filename,
        "content_type": file.content_type,
        "size": file_size,
        "description": description,
        "category": category,
        "message": "File uploaded successfully"
    }

错误处理与调试

参数验证错误处理

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# 自定义异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """处理请求验证错误"""
    errors = []
    for error in exc.errors():
        errors.append({
            "loc": error['loc'],
            "msg": error['msg'],
            "type": error['type'],
            "input": error.get('input')
        })
    
    return JSONResponse(
        status_code=422,
        content={
            "detail": "参数验证失败",
            "errors": errors,
            "message": "请检查请求参数格式和值"
        }
    )

class StrictItem(BaseModel):
    """严格验证的项目模型"""
    name: str = Field(
        ..., 
        min_length=5, 
        max_length=50, 
        description="项目名称,5-50字符"
    )
    price: float = Field(
        ..., 
        gt=0, 
        lt=10000, 
        description="价格,0-10000"
    )
    category: str = Field(
        ..., 
        regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$", 
        description="分类名,字母数字下划线"
    )

@app.post("/strict-items/")
def create_strict_item(item: StrictItem):
    """
    创建严格验证的项目
    演示错误处理
    """
    return item

# 自定义业务逻辑错误
@app.get("/items/{item_id}")
def get_item_with_validation(
    item_id: int = Field(..., gt=0, description="项目ID必须大于0")
):
    """
    获取项目(带业务验证)
    """
    # 模拟业务逻辑验证
    if item_id > 10000:
        raise HTTPException(
            status_code=400,
            detail=f"项目ID {item_id} 超出有效范围"
        )
    
    # 模拟数据不存在
    if item_id == 9999:
        raise HTTPException(
            status_code=404,
            detail=f"项目 {item_id} 不存在"
        )
    
    return {"item_id": item_id, "name": f"Item {item_id}"}

调试技巧

from fastapi import FastAPI, Query, Path, Body
from typing import Annotated, Optional
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = DebugApp()  # Note: Using DebugApp instead of FastAPI for demonstration

class DebugApp(FastAPI):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
    
    def openapi(self):
        # 重写OpenAPI文档生成,添加调试信息
        if not self.debug:
            return super().openapi()
        
        # 添加调试信息到OpenAPI文档
        openapi_schema = super().openapi()
        if openapi_schema:
            openapi_schema["info"]["x-debug-mode"] = True
        return openapi_schema

app = FastAPI(debug=True)

@app.get("/debug-parameters/")
def debug_parameters(
    item_id: Annotated[
        int,
        Path(title="项目ID", description="路径参数调试")
    ],
    q: Annotated[
        Optional[str],
        Query(description="查询参数调试")
    ] = None,
    debug: Annotated[
        bool,
        Query(description="调试模式开关")
    ] = False
):
    """
    参数调试端点
    用于调试参数接收和处理
    """
    # 记录参数信息
    param_info = {
        "item_id": {"value": item_id, "type": type(item_id).__name__},
        "q": {"value": q, "type": type(q).__name__ if q is not None else "NoneType"},
        "debug": {"value": debug, "type": type(debug).__name__}
    }
    
    logger.info(f"Received parameters: {param_info}")
    
    result = {
        "received_params": param_info,
        "processed": True
    }
    
    if debug:
        result["debug_info"] = {
            "endpoint": "/debug-parameters/",
            "method": "GET",
            "expected_types": {
                "item_id": "int",
                "q": "Optional[str]",
                "debug": "bool"
            }
        }
    
    return result

性能优化

参数处理优化

from fastapi import FastAPI
from pydantic import BaseModel, Field, validator
from typing import Optional, List
import cProfile
import pstats
from functools import wraps

app = FastAPI()

# 使用缓存优化频繁的参数验证
from functools import lru_cache

@lru_cache(maxsize=128)
def validate_category_cached(category: str) -> bool:
    """缓存分类验证结果"""
    allowed_categories = frozenset([
        'electronics', 'books', 'clothing', 'home', 'sports', 
        'beauty', 'toys', 'automotive', 'garden', 'tools'
    ])
    return category.lower() in allowed_categories

class OptimizedProduct(BaseModel):
    """优化的产品模型"""
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    category: str
    tags: List[str] = []
    
    @validator('category')
    def validate_category_optimized(cls, v):
        """使用缓存的分类验证"""
        if not validate_category_cached(v.lower()):
            raise ValueError(f'无效的分类: {v}')
        return v.lower()
    
    class Config:
        # 性能优化配置
        validate_assignment = False  # 禁用赋值验证以提高性能
        arbitrary_types_allowed = True

@app.post("/optimized-products/")
def create_optimized_product(product: OptimizedProduct):
    """
    创建优化的产品
    演示性能优化技巧
    """
    return product

# 参数批量处理
from typing import List as TypingList

@app.post("/batch-create/")
def batch_create_products(products: TypingList[OptimizedProduct]):
    """
    批量创建产品
    优化批量参数处理
    """
    # 批量验证和处理
    processed_products = []
    for i, product in enumerate(products):
        product_dict = product.model_dump()
        product_dict['batch_id'] = i
        processed_products.append(product_dict)
    
    return {
        "created_count": len(processed_products),
        "products": processed_products
    }

验证器性能优化

import re
from fastapi import FastAPI
from pydantic import BaseModel, Field, validator
from typing import Optional

app = FastAPI()

# 预编译正则表达式以提高性能
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
PHONE_REGEX = re.compile(r"^\+?1?-?\.?\s?\(?(\d{3})\)?[\s\.-]?(\d{3})[\s\.-]?(\d{4})$")

class PerformanceOptimizedUser(BaseModel):
    """性能优化的用户模型"""
    email: str = Field(..., description="邮箱地址")
    phone: Optional[str] = Field(None, description="电话号码")
    
    @validator('email')
    def validate_email_performance(cls, v):
        """性能优化的邮箱验证"""
        if not EMAIL_REGEX.match(v):
            raise ValueError('无效的邮箱格式')
        return v.lower()
    
    @validator('phone', pre=True)
    def validate_phone_performance(cls, v):
        """性能优化的电话验证"""
        if v is None:
            return v
        # 移除所有非数字字符进行验证
        digits_only = ''.join(filter(str.isdigit, v))
        if len(digits_only) < 10 or len(digits_only) > 15:
            raise ValueError('电话号码长度无效')
        return v

@app.post("/perf-users/")
def create_performance_user(user: PerformanceOptimizedUser):
    """
    创建性能优化的用户
    演示验证器性能优化
    """
    return user

相关教程

合理使用Pydantic模型验证可以大大减少手动验证代码,提高开发效率和代码质量。建议为每个API端点明确定义参数类型和验证规则。 对于高频API调用,考虑使用缓存验证结果、预编译正则表达式和批量处理来优化参数验证性能。

总结

FastAPI的参数处理系统提供了强大而灵活的功能:

  1. 多种参数类型:支持路径参数、查询参数、请求体等多种参数类型
  2. 自动验证:基于类型提示的自动数据验证
  3. 灵活约束:通过Pydantic Field和Query提供丰富的约束选项
  4. 错误处理:完善的错误处理和调试支持
  5. 性能优化:支持缓存、批量处理等性能优化技术

通过合理使用这些功能,您可以构建健壮、高效且易于维护的API应用。掌握参数处理是成为FastAPI专家的重要一步。