第一章：基础架构重构
1.1 传统模式的问题诊断
PYTHON

典型问题代码

@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()
缺陷分析：

线性增长的维护成本（每新增条件需修改代码）
条件组合导致测试用例爆炸式增长
无法实现动态vb.net教程C#教程python教程SQL教程access 2010教程字段扩展
1.2 现代化解决方案
PYTHON

# 声明式过滤配置
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_)
    ]

技术亮点：

类型安全验证（自动过滤非法类型参数）
配置与逻辑解耦（新增条件只需修改配置）
支持动态字段扩展
第二章：动态查询构建
2.1 策略模式实现
PYTHON

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"]))

2.2 复合查询构建
PYTHON

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))

第三章：安全与验证
3.1 参数验证模型
PYTHON

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

3.2 SQL注入防护
PYTHON

# 不安全做法（绝对禁止！）
query.filter(f"price > {user_input}")

# 安全做法
from sqlalchemy import text

query.filter(text("price > :min_price")).params(min_price=user_input)

第四章：性能优化
4.1 索引策略
SQL

-- 复合索引
CREATE INDEX idx_items_search ON items (category, price DESC);

-- 函数索引
CREATE INDEX idx_name_lower ON items (LOWER(name));

4.2 分页优化对比
PYTHON

# 传统分页（性能随offset增大线性下降）
query.offset((page - 1) * size).limit(size)

# 游标分页（恒定时间查询）
last_id = request.query_params.get("last_id")
query.filter(Item.id > last_id).limit(size)

第五章：错误处理
5.1 统一错误响应
PYTHON

@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()
            ]
        }
    )

5.2 常见错误速查

Q2：哪种分页方式更适合大数据量场景？

Offset分页
游标分页
随机分页
Q3：如何验证价格区间的有效性？

在前端进行验证
使用Pydantic自定义校验器
在数据库添加CHECK约束

来源：https://blog.cmdragon.cn/posts/63d68d803116/