我正在开发一个执行报告类型的应用程序，我经常需要使用 JSON 指定的过滤器，将其转换为 SQLAlchemy 查询，然后将结果发送回浏览器（例如过滤器start_date: 1234, end_date: 5678, widget_ids: [1, 2, 3]需要转换进入查询… WHERE start_date >= 1234 AND end_date <= 5678 AND widget_id in (1, 2, 3)）。
是否有工具可以自动进行这种转换（例如，使用 Django 的 ORM: 之类的后缀start_date__ge: 1234, end_date__le: 5678, widget_id__in: [1, 2, 3]）？
显然，这样的方案会对安全性和性能产生影响……但我想知道在我自己构建之前是否存在类似的东西。
编辑：我意识到我可以构建自己的东西，但我特别想知道是否有现有的工具/库，所以我不需要重新发明轮子。
我有一些可能感兴趣的简单代码（来源）：
def create_attr_filter(request, mapped_class):
"""Create an ``and_`` SQLAlchemy filter (a ClauseList object) based
on the request params (``queryable``, ``eq``, ``ne``, ...).
Arguments:
request
the request.
mapped_class
the SQLAlchemy mapped class.
"""
mapping = {
'eq' : '__eq__',
'ne' : '__ne__',
'lt' : '__lt__',
'lte' : '__le__',
'gt' : '__gt__',
'gte' : '__ge__',
'like' : 'like',
'ilike': 'ilike'
}
filters = []
if 'queryable' in request.params:
queryable = request.params['queryable'].split(',')
for k in request.params:
if len(request.params[k]) <= 0 or '__' not in k:
continue
col, op = k.split("__")
if col not in queryable or op not in mapping.keys():
continue
column = getattr(mapped_class, col)
f = getattr(column, mapping[op])(request.params[k])
filters.append(f)
return and_(*filters) if len(filters) > 0 else None
我写了类似的东西。我称之为 Python 代理。它提供了一个 Javascript API 来通过 JSON 对 Python 进行 RPC。
这是我的开源项目的一部分。
相关文件是：
http://code.google.com/p/pycopia/source/browse/trunk/WWW/pycopia/WWW/json.py#135
Javascript方面：
http://code.google.com/p/pycopia/source/browse/trunk/WWW/media/js/proxy.js
用法：
http://code.google.com/p/pycopia/source/browse/trunk/storage/pycopia/db/webservice.py
示例配置：
http://code.google.com/p/pycopia/source/browse/trunk/storage/etc/storage.conf.example
但是，它是完整框架的一部分，可惜没有很好的文档记录。需要最新的 Linux 平台和一些设置。
但也许你可以从中得到一些想法。
下面是一些从 CLI 样式 (argv) 字符串构建查询的其他代码。它进行动态运算符选择。
def _get_query(self, argv):
mapper = models.class_mapper(self._obj)
args, kwargs = _query_args(argv[1:], self._environ)
q = _session.query(self._obj)
if args:
grps, left = divmod(len(args), 3)
if grps:
for name, op, val in _by_three(args[:grps*3]):
col = getattr(self._obj, name)
opm = {"=": col.__eq__,
">": col.__gt__,
"<": col.__lt__,
"match": col.match,
"contains": col.contains,
"in": col.in_,
"like": col.like}.get(op)
if opm:
if op == "like":
val = val.replace("*", "%")
val = val.replace(".", "_")
if "%" not in val:
val = "%" + val + "%"
if op == "in":
val = val.split(",")
q = q.filter(opm(val))
for name in args[grps*3:]:
if name.startswith("="):
q = q.order_by(name[1:])
if kwargs:
for name, value in kwargs.items():
col = getattr(self._obj, name)
value = CLI.clieval(value)
q = q.filter(col.__eq__(value))
return q
这也可能让你开始。
查看您的示例，您的JSON字符串实际上并不包含operators，因此不清楚它是否是==, >=, <=, IN, etc. 您的JSON字符串可以轻松转换为dict.
对于你的简单情况have ==，你可以简单地使用filter_by提供一个命名的字典：
query_dict = {'name': 'parent-2', 'description': 'test', }
query = query.filter_by(**query_dict)
同样，您可以使用SQLAlchemygetattr中的回答来构建查询 - 从 dict 动态构建查询过滤器，但这仍然需要知道operator条件的。在下面的代码中仅like使用：
q = session.query(myClass)
for attr, value in web_dict.items():
q = q.filter(getattr(myClass, attr).like("%%%s%%" % value))
如果您需要动态指定运算符，则可以进一步使用getattr：
klass, attr, oper, value = MyClass, "startDate", "__ge__", 1234
q = q.filter(getattr(getattr(klass, attr), oper)(value))