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)

# 传统分页（性能随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()
            ]
        }
    )