FastAPI+SQLAlchemy集成ERM时动态查询参数过滤的实现疑问

原方案的问题 #

你当前用静态Pydantic Schema的方案不具备可扩展性，核心问题有两个：

1. 硬编码的RecordQuerySchema无法适配ERM动态新增的主/扩展选项，每次新增过滤字段都要修改代码，违背了"无需手动迁移"的需求；
2. 在Pydantic模型字段中使用Depends()是错误用法，Depends仅用于路由函数的参数依赖注入，放在模型里会导致FastAPI解析查询参数时出现必填参数异常。

可行的解决思路 #

1. 动态解析查询参数，基于过滤规则做合法性校验 #

放弃静态Schema，直接从Request对象中获取所有查询参数，结合FilterCategory表中当前分类（category_id）允许的过滤字段，对参数做合法性校验：

• 先查询当前category_id对应的所有FilterCategory记录，得到允许的过滤选项集合；
• 遍历所有查询参数，仅保留在允许集合内的参数，非法参数直接忽略或抛出异常。

2. 动态构建SQLAlchemy查询条件 #

区分主选项和扩展选项，分别构建过滤逻辑：

• 主选项：直接匹配Table模型的字段，根据参数名中的操作符（如_gt/_lt/_eq）生成对应的SQLAlchemy比较表达式；
• 扩展选项：需要关联ExtraOption和TableExtraOption表，先通过参数名找到对应的extra_option_id，再根据参数值过滤TableExtraOption.value。

3. 适配ERM的动态同步需求 #

通过监听ERM的事件（如Webhook），当ERM新增ExtraOption或FilterCategory时，自动同步到本服务的数据库，确保查询时能拿到最新的允许过滤字段，完全无需手动执行数据库迁移。

代码示例 #

from fastapi import FastAPI, Request, Depends
from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
import uuid as uuid_pkg

# 假设已定义好ORM模型：Table, ExtraOption, TableExtraOption, FilterCategory

app = FastAPI()

async def get_db() -> AsyncSession:
    # 此处实现你的数据库会话获取逻辑
    pass

@app.get("/records")
async def get_records(
    category_id: uuid_pkg.UUID,
    request: Request,
    db: AsyncSession = Depends(get_db)
):
    # 1. 获取当前分类允许的所有过滤选项
    filter_categories = await db.execute(
        select(FilterCategory).where(FilterCategory.category_id == category_id)
    )
    allowed_fields = {fc.option for fc in filter_categories.scalars().all()}

    # 2. 解析查询参数
    query_params = dict(request.query_params)
    conditions = []

    for param_name, param_value in query_params.items():
        # 解析字段名和操作符（如 length_gt -> 字段length，操作符gt）
        if "_" in param_name:
            field_name, op = param_name.rsplit("_", 1)
        else:
            field_name = param_name
            op = "eq"

        # 跳过不在允许列表中的字段
        if field_name not in allowed_fields:
            continue

        # 处理主选项过滤
        if hasattr(Table, field_name):
            model_field = getattr(Table, field_name)
            # 根据操作符生成条件
            if op == "gt":
                conditions.append(model_field > float(param_value))
            elif op == "lt":
                conditions.append(model_field < float(param_value))
            elif op == "eq":
                conditions.append(model_field == param_value)
            # 可按需扩展其他操作符（如_ne, _in等）
        
        # 处理扩展选项过滤
        else:
            # 先找到对应的ExtraOption记录
            extra_option = await db.execute(
                select(ExtraOption).where(ExtraOption.extra_option == field_name)
            )
            extra_option = extra_option.scalar_one_or_none()
            if not extra_option:
                continue
            
            # 构建关联查询条件
            conditions.append(and_(
                Table.id == TableExtraOption.table_id,
                TableExtraOption.extra_option_id == extra_option.id,
                # 根据操作符匹配value，需根据实际数据类型调整转换逻辑
                getattr(TableExtraOption.value, op)(param_value)
            ))

    # 3. 执行查询并返回结果
    query = select(Table).where(*conditions)
    records = await db.execute(query)
    return records.scalars().all()

补充说明 #

• 操作符的规则可以根据需求自定义，比如支持_in（多值用逗号分隔）、_contains等，只需在解析时对应处理；
• 扩展选项的value类型可能多样（字符串/数字/布尔），可以在ExtraOption表中添加data_type字段，同步时保存类型信息，查询时做对应类型转换；
• 若需要分页、排序功能，也可以用同样的动态方式处理，基于FilterCategory中的order字段生成排序逻辑。

内容的提问来源于stack exchange，提问作者Konstantinos