#FastAPI路径参数与查询参数详解
📂 所属阶段:第一阶段 — 快速筑基(基础篇)
🔗 相关章节:FastAPI简介与优势 · 环境搭建 · Pydantic综合指南
#目录
#参数处理概述
在Web开发中,参数处理是构建API的核心环节。FastAPI提供了强大而灵活的参数处理机制,基于Python类型提示实现自动验证和转换。
#FastAPI参数类型
FastAPI支持多种参数类型:
- 路径参数 (Path Parameters) - 从URL路径中提取
- 查询参数 (Query Parameters) - 从URL查询字符串中提取
- 请求体参数 (Request Body) - 从请求体中提取
- Cookie参数 (Cookie Parameters) - 从Cookie中提取
- 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#相关教程
#总结
FastAPI的参数处理系统提供了强大而灵活的功能:
- 多种参数类型:支持路径参数、查询参数、请求体等多种参数类型
- 自动验证:基于类型提示的自动数据验证
- 灵活约束:通过Pydantic Field和Query提供丰富的约束选项
- 错误处理:完善的错误处理和调试支持
- 性能优化:支持缓存、批量处理等性能优化技术
通过合理使用这些功能,您可以构建健壮、高效且易于维护的API应用。掌握参数处理是成为FastAPI专家的重要一步。

