You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

是否存在将类MongoDB查询字典转SQLAlchemy查询的工具?

解决方案:Mongo风格查询字典转SQLAlchemy查询的工具与替代方案

一、已有成熟工具库

直接用sqlalchemy-mongoquery就能实现你要的功能,它专门用来将MongoDB风格的查询字典转换成SQLAlchemy的查询过滤条件,完美匹配你给出的场景。

示例代码(对应你提供的查询需求):

from sqlalchemy_mongoquery import QueryBuilder
from datetime import datetime, timedelta

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

builder = QueryBuilder(User)
query = session.query(User).filter(builder.build(query_dict))
results = query.all()

这个库支持绝大多数MongoDB查询操作符,包括嵌套的$and、$or、$not等复杂逻辑,完全满足复杂查询的转换需求。

另外,sqlalchemy-filters也能实现类似效果,它支持从字典生成过滤条件,虽然语法和Mongo不完全一致,但灵活性很高,适合需要自定义规则的场景。

二、支持字典式查询的SQLAlchemy替代ORM

如果不想在SQLAlchemy上做转换,也可以直接选择本身就支持类似PyMongo字典查询风格的ORM:

  • Tortoise-ORM:异步ORM,支持通过双下划线语法模拟Mongo查询逻辑,写法简洁:
    results = await User.filter(
        name__contains='John',
        age__in=[25,30,35],
        created_at__gte=one_week_ago
    ).all()
    
  • Peewee:轻量级ORM,支持通过字典或自定义函数构造查询条件,比如:
    from peewee import Expression
    def dict_to_expr(model, query_dict):
        exprs = []
        for k, v in query_dict.items():
            field = getattr(model, k)
            if isinstance(v, dict):
                for op, val in v.items():
                    if op == '$in':
                        exprs.append(field << val)
                    elif op == '$gte':
                        exprs.append(field >= val)
                    elif op == '$regex':
                        exprs.append(field.contains(val))
        return Expression(*exprs, operator='AND')
    query = User.select().where(dict_to_expr(User, query_dict))
    

三、自行实现转换函数(无第三方依赖)

如果需要完全自定义逻辑,也可以自己写一个简单的转换函数,核心是把Mongo操作符映射到SQLAlchemy的对应方法:

from sqlalchemy import and_

def mongo_dict_to_sqlalchemy_filter(model, query_dict):
    filters = []
    for field_name, conditions in query_dict.items():
        field = getattr(model, field_name)
        if isinstance(conditions, dict):
            for op, value in conditions.items():
                if op == '$regex':
                    filters.append(field.like(f'%{value}%'))
                elif op == '$in':
                    filters.append(field.in_(value))
                elif op == '$gte':
                    filters.append(field >= value)
                # 可扩展支持$lt、$or、$not等更多操作符
        else:
            filters.append(field == conditions)
    return and_(*filters)

# 使用示例
query_filter = mongo_dict_to_sqlalchemy_filter(User, query_dict)
results = session.query(User).filter(query_filter).all()

内容的提问来源于stack exchange,提问作者gebbissimo

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.30 15:23:09