是否存在将类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
相关产品推荐
相关产品推荐

