FastAPI复杂查询终极指南：告别if-else的现代化过滤架构
title: FastAPI复杂查询终极指南：告别if-else的现代化过滤架构
# 典型问题代码
@app.get("/items")
def get_items(name: str = None, min_price: float = None, ...):
query = Item.query
if name: query = query.filter_by(name=name)
if min_price: query = query.filter(price >= min_price)
# 每新增一个条件需增加一个if分支
return query.all()
# 声明式过滤配置
filter_config = {
"name": (lambda v: Item.name == v, str),
"price_gte": (lambda v: Item.price >= v, float),
"category_in": (lambda v: Item.category.in_(v), list)
}
def build_filters(params: dict):
return [
logic(param) for field, (logic, type_) in filter_config.items()
if (param := params.get(field)) is not None
and isinstance(param, type_)
]
class FilterStrategy:
_strategies = {}
@classmethod
def register(cls, name):
def decorator(func):
cls._strategies[name] = func
return func
return decorator
@classmethod
def apply(cls, query, params):
for param, value in params.items():
if strategy := cls._strategies.get(param):
query = strategy(query, value)
return query
@FilterStrategy.register("name_contains")
def _(query, value):
return query.filter(Item.name.ilike(f"%{value}%"))
@FilterStrategy.register("price_range")
def _(query, value: dict):
return query.filter(Item.price.between(value["min"], value["max"]))
from sqlalchemy import and_, or_
def build_composite_filter(filters: list, logic_gate=and_):
return logic_gate(*[filt for filt in filters if filt is not None])
# 使用示例
filters = [
Item.price >= 100,
or_(Item.category == "electronics", Item.category == "furniture")
]
query = session.query(Item).filter(build_composite_filter(filters))
from pydantic import BaseModel, conlist, confloat
class AdvancedFilter(BaseModel):
search_term: Optional[str] = Field(max_length=50)
price_range: Optional[dict] = Field(
regex="^{min:\d+,max:\d+}$",
example={"min": 100, "max": 500}
)
sort_by: Optional[str] = Field(regex="^(name|price)(_desc)?$")
@validator("price_range")
def validate_price_range(cls, v):
if v["min"] > v["max"]:
raise ValueError("Min price exceeds max")
return v
# 不安全做法（绝对禁止！）
query.filter(f"price > {user_input}")
# 安全做法
from sqlalchemy import text
query.filter(text("price > :min_price")).params(min_price=user_input)
-- 复合索引
CREATE INDEX idx_items_search ON items (category, price DESC);
-- 函数索引
CREATE INDEX idx_name_lower ON items (LOWER(name));
# 传统分页（性能随offset增大线性下降）
query.offset((page - 1) * size).limit(size)
# 游标分页（恒定时间查询）
last_id = request.query_params.get("last_id")
query.filter(Item.id > last_id).limit(size)
@app.exception_handler(ValidationError)
async def handle_validation_error(request, exc):
return JSONResponse(
status_code=422,
content={
"detail": "参数校验失败",
"errors": [
f"{'.'.join(map(str, e['loc']))}: {e['msg']}"
for e in exc.errors()
]
}
)
| 错误码 | 场景 | 解决方案 |
|---|---|---|
| 422 | 参数类型错误 | 检查Pydantic模型约束条件 |
| 500 | 无效排序字段 | 添加字段白名单验证 |
| 429 | 复杂查询频率过高 | 实现基于查询复杂度的限流策略 |
评论
发表评论