动态sqlalchemy查询

Question

我有2个sqlalchemy查询...

dummy = DBSession().query(Dummy).filter(Dummy.c_name) == "foo")

dummy2 = DBSession().query(Dummy2).filter(Dummy2.c_name == "foo").filter(Dummy2.f_name == "bar")

而且我有某种组合了两者的通用函数...

def generic(Object, c_name, f_name): 
    dummy = DBSession().query(Object).filter(Object.c_name == c_name).filter(Object.f_name == f_name)

一般处理此问题的最佳方法是什么，比如说f_name在Dummy2表中不存在或不可查询？

总结我的问题：

我如何创建一个通用的sqlalchemy查询，该查询可以查询任何给定的表，在某些情况下，我要查询的属性根据给定的对象而有所不同。

我想我需要某种反思...也许吗？还是* args / ** kwargs ...我不知道...帮助吗？

Answer 1

您可能应该将它们打包在单独的帮助器函数中以实现可重用性，但是这是我的处理方法：

check_fields = {'c_name': 'exists', 'f_name': 'doesnt_exist'}
existing_fields = {}
# check if the fields exist in the table
for field in check_fields:
    if field in given_table.c:
        existing_fields.update({field: check_fields.get(field)})

# construct the query based on existing fields
query = given_table.select()
for k, v in existing_fields.items():
    query = query.where(getattr(given_table.c, k) == v)

然后我使用session.execute（query）获得结果。这是一个不使用execute的类似答案：How to get query in sqlalchemyORM также

注意： 所有属性将与AND链接在一起