Dynamically add filter to SQLAlchemy TextClause

Question

Assume I have a SQLAlchemy table which looks like:

class Country:
    name = VARCHAR
    population = INTEGER
    continent = VARCHAR
    num_states = INTEGER

My application allow seeing name and population for all Countries. So I have a TextClause which looks like

"select name, population from Country"

I allow raw queries in my application so I don't have option to change this to selectable.

At runtime, I want to allow my users to choose a field name and put a field value on which I want to allow filtering. eg: User can say I only want to see name and population for countries where Continent is Asia. So I dynamically want to add the filter

.where(Country.c.continent == 'Asia')

But I can't add .where to a TextClause.

Similarly, my user may choose to see name and population for countries where num_states is greater than 10. So I dynamically want to add the filter

.where(Country.c.num_states > 10)

But again I can't add .where to a TextClause.

What are the options I have to solve this problem?

Could subquery help here in any way?

Answer 1

Please add a filter based on the conditions. filter is used for adding where conditions in sqlalchemy.

Country.query.filter(Country.num_states > 10).all()

You can also do this:

query = Country.query.filter(Country.continent == 'Asia')
if user_input == 'states':
    query = query.filter(Country.num_states > 10)
query = query.all()

Answer 2

This is not doable in a general sense without parsing the query. In relational algebra terms, the user applies projection and selection operations to a table, and you want to apply selection operations to it. Since the user can apply arbitrary projections (e.g. user supplies SELECT id FROM table), you are not guaranteed to be able to always apply your filters on top, so you have to apply your filters before the user does. That means you need to rewrite it to SELECT id FROM (some subquery), which requires parsing the user's query.

However, we can sort of cheat depending on the database that you are using, by having the database engine do the parsing for you. The way to do this is with CTEs, by basically shadowing the table name with a CTE.

Using your example, it looks like the following. User supplies query

SELECT name, population FROM country;

You shadow country with a CTE:

WITH country AS (
  SELECT * FROM country
  WHERE continent = 'Asia'
) SELECT name, population FROM country;

Unfortunately, because of the way SQLAlchemy's CTE support works, it is tough to get it to generate a CTE for a TextClause. The solution is to basically generate the string yourself, using a custom compilation extension, something like this:

class WrappedQuery(Executable, ClauseElement):
    def __init__(self, name, outer, inner):
        self.name = name
        self.outer = outer
        self.inner = inner

@compiles(WrappedQuery)
def compile_wrapped_query(element, compiler, **kwargs):
    return "WITH {} AS ({}) {}".format(
        element.name,
        compiler.process(element.outer),
        compiler.process(element.inner))

c = Country.__table__
cte = select(["*"]).select_from(c).where(c.c.continent == "Asia")
query = WrappedQuery("country", cte, text("SELECT name, population FROM country"))
session.execute(query)

From my tests, this only works in PostgreSQL. SQLite and SQL Server both treat it as recursive instead of shadowing, and MySQL does not support CTEs.