Dynamic sqlalchemy query

Question

I have 2 sqlalchemy queries...

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

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

And I have some sort of made up a generic function that combines the two...


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

What's the best way to generically handle this, say if f_name doesn't exist or is not queryable in the Dummy2 table?

To summarise my question:

How do I create a general sqlalchemy query which can query any given table, where in some cases the attributes I am querying varies based on the given object.

I think I need some sort of reflection...maybe? or *args / **kwargs ... I dunno... help?

c8999c 3f964f64 · Answer 1 · 2019-11-27T16:34:35.843

you should probably pack these in separate helper functions for reusability but here's how I approached this:

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()
if existing_fields:
    for k, v in existing_fields.items():
        query = query.where(getattr(given_table.c, k) == v)

I'm then using session.execute(query) to get the results. Here's a similar answer that doesn't use execute: How to get query in sqlalchemyORM также

Note: All the attributes will be chained with AND

Dynamic sqlalchemy query

1 Answers1