Как отредактировать этот sqlalchemy фильтр запросов «по умолчанию», чтобы сделать его приоритетным

Question

РЕДАКТИРОВАТЬ: Я думал, что порядок в условиях WHERE в SQL имеет значение, и поэтому я попросил решение проблемы.Поэтому я получил свой ответ.

Я пытаюсь установить фильтр запросов по умолчанию в модели sqlalchemy, которая выполняется каждый раз, когда запрос вызывается.

Но проблема в том, что он добавляет фильтр приконец запроса, который нарушает мой SQL.

Возможно ли сделать его первым фильтром в запросе?

Я следовал этому рецепту: https://github.com/sqlalchemy/sqlalchemy/wiki/FilteredQuery

Skeletonиз рецепта:

# 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 ставится в конце моего SQL-запроса.

Answer 1

Я думаю, вам следует добавить оператор break при добавлении фильтра в запрос, в противном случае вы рискуете выполнить фильтрацию по одному и тому же условию несколько раз.

Глядя на внутренние компоненты метода фильтра, я вижу следующий код, который нужно выполнить для получения списка условий или критериев:

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

Это означает, что вы можете добавить условие с помощью следующего кода:

@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