SQLAlchemy 中实现 MongoDB 风格字典查询的实践方案

本文探讨如何在 sqlalchemy 中模拟 mongodb 的字典式复杂查询（如 `$regex`、`$in`、`$gte`），分析原生支持限制，提供轻量级字典到 orm 查询的转换思路，并指出成熟库的缺失现状与实用替代策略。

SQLAlchemy 本身不原生支持 MongoDB 风格的嵌套字典查询语法（如 {'name': {'$regex': 'John'}, 'age': {'$in': [25,30,35]}}），其设计哲学强调显式性、类型安全与 SQL 可控性——这与 MongoDB 的动态文档查询范式存在根本差异。但并不意味着无法实现类似能力：关键在于将字典结构映射为 SQLAlchemy 可识别的表达式对象（ClauseElement），再组合进 .filter()。

以下是一个简洁、可扩展的转换函数示例，支持常见操作符：

from sqlalchemy import and_, or_, not_
from sqlalchemy.sql import operators
from sqlalchemy.orm import Query

def dict_to_sqlalchemy_filter(model, query_dict):
    """
    将 MongoDB 风格字典转换为 SQLAlchemy filter 表达式
    支持: $eq, $ne, $in, $nin, $like, $ilike, $gt, $gte, $lt, $lte, $regex (→ ilike), $and, $or, $not
    """
    filters = []
    for key, value in query_dict.items():
        if isinstance(value, dict) and len(value) == 1:
            op, operand = next(iter(value.items()))
            attr = getattr(model, key)
            if op == "$eq":
                filters.append(attr == operand)
            elif op == "$ne":
                filters.append(attr != operand)
            elif op == "$in":
                filters.append(attr.in_(operand))
            elif op == "$nin":
                filters.append(~attr.in_(operand))
            elif op in ("$gt", "$gte", "$lt", "$lte"):
                op_map = {"$gt": operators.gt, "$gte": operators.ge,
                          "$lt": operators.lt, "$lte": operators.le}
                filters.append(op_map[op](attr, operand))
            elif op in ("$like", "$ilike"):
                filters.append(getattr(attr, op)(f"%{operand}%"))
            elif op == "$regex":
                # 简单兼容：转为不区分大小写的模糊匹配（PostgreSQL/SQLite）
                filters.append(attr.ilike(f"%{operand}%"))
        elif key.startswith("$"):
            # 处理顶层逻辑操作符：$and, $or, $not
            if key == "$and":
                sub_filters = [dict_to_sqlalchemy_filter(model, subq) for subq in value]
                filters.append(and_(*sub_filters))
            elif key == "$or":
                sub_filters = [dict_to_sqlalchemy_filter(model, subq) for subq in value]
                filters.append(or_(*sub_filters))
            elif key == "$not":
                sub_filter = dict_to_sqlalchemy_filter(model, value)
                filters.append(not_(sub_filter))
        else:
            # 默认行为：精确匹配（$eq 语义）
            filters.append(getattr(model, key) == value)
    return and_(*filters) if filters else True

# 使用示例
one_week_ago = datetime.utcnow() - timedelta(days=7)
query_dict = {
    "name": {"$regex": "John"},
    "age": {"$in": [25, 30, 35]},
    "created_at": {"$gte": one_week_ago},
}

# 构建查询
query = session.query(User).filter(dict_to_sqlalchemy_filter(User, query_dict))
results = query.all()

⚠️ 注意事项与权衡：

• 安全性：该方案仍需确保 query_dict 来源可信，避免注入风险（如用户可控字段名或恶意操作符）；
• 数据库兼容性：$regex 映射为 ilike 是跨库妥协；若需真正正则（如 PostgreSQL 的 ~），需按方言定制；
• 性能：过度抽象可能掩盖 N+1 或低效查询问题，建议配合 explain() 分析执行计划；
• 生态现状：截至 SQLAlchemy 2.0，尚无被广泛采用的第三方库（如 sqlalchemy-mongo-query）提供开箱即用的完整 MongoDB 语法支持。社区更倾向使用原生表达式或转向异构方案（如通过 SQLModel + pydantic 做中间层校验，或在 API 层统一解析后分发至不同 ORM）。

✅ 总结：虽无“银弹”库，但通过封装 dict → ClauseElement 转换逻辑，即可在保持 SQLAlchemy 安全性与可维护性的前提下，获得接近 MongoDB 的开发体验。核心原则是——让抽象服务于清晰，而非掩盖复杂性。