CODEWITHSUNDEEP
pythonsqlalchemy
mikro45
mikro45

Reputation: 63

How to edit this sqlalchemy "default" query filter to make it priority

EDIT: I thought that order in WHERE conditions in SQL matter and so was the reason for me asking for the solution to the problem. Therefore, I got my answer.

I am trying to set default query filter in sqlalchemy model, that executes every time the query is called.

But the problem is that it adds the filter at the end of the query, which breaks my SQL.

Is it possible to make it first filter in query?

I was following this recipe: https://github.com/sqlalchemy/sqlalchemy/wiki/FilteredQuery

Skeleton from the recipe:

# see the linked web page, for full code

@event.listens_for(Query, "before_compile", retval=True)
def before_compile(query):
    for ent in query.column_descriptions:
        entity = ent['entity']
        if entity is None:
            continue
        insp = inspect(ent['entity'])
        mapper = getattr(insp, 'mapper', None)
        if mapper and issubclass(mapper.class_, HasPrivate):
            query = query.enable_assertions(False).filter(
                ent['entity'].public == True)

    return query


# this is my filter
query = query.enable_assertions(False).filter(
    ent['entity'].id != ent['entity'].other_id)

ent['entity'].id != ent['entity'].other_id is put at the end of my SQL query.

Upvotes: 0

Views: 472

Answers (1)

Ruben Helsloot
Ruben Helsloot

Reputation: 13129

I think you should add a break statement when you add the filter to the query, otherwise you risk filtering on the same condition multiple times.

Looking at the internals of the filter method, I see the following code to execute for a list of conditions or criterions:

for criterion in list(criterion):
    criterion = expression._expression_literal_as_text(criterion)

    criterion = self._adapt_clause(criterion, True, True)

    if self._criterion is not None:
        self._criterion = self._criterion & criterion
    else:
        self._criterion = criterion

This implies that you can prepend the condition using the following code:

@event.listens_for(Query, "before_compile", retval=True)
def before_compile(query):
    criterion = query._adapt_clause(criterion, True, True)
    for ent in query.column_descriptions:
        entity = ent['entity']
        if entity is None:
            continue
        insp = inspect(ent['entity'])
        mapper = getattr(insp, 'mapper', None)
        if mapper and issubclass(mapper.class_, HasPrivate):
            if query._criterion is not None:
               query._criterion = criterion & query._criterion
            else:
               query._criterion = criterion
            break
    return query

Upvotes: 1

Related Questions