SQLAlchemy - объединяет текстовый запрос с фильтром

Question

Я использую SA 0.6.6 , Python 2.66 и Postgres 8.3 .

У меня есть определенные запросы, которые требуют довольно сложной проверки безопасности, которая может быть обработана с помощью запроса WITH RECURSIVE. Я пытаюсь объединить текстовый запрос с объектом запроса, чтобы я мог применять фильтры по мере необходимости.

Первоначально я думал создать текстовый запрос как подзапрос, а затем объединить его с запросом пользователя и фильтрами. К сожалению, это не работает.

subquery = session.query(sharedFilterAlias).\
                   from_statement(sharedFilterQuery).subquery()

Это приводит к этой ошибке:

AttributeError: 'Annotated_TextClause' object has no attribute 'alias'

Есть ли способ объединить текстовый запрос с объектом запроса SQLAlchemy?

Answer 1

Через некоторое время без ответа я отправил сообщение в SA Google Group , где Майкл Байер сам поставил меня в правильном направлении.

Ответ - превратить мой текстовый запрос в текстовое предложение SA. Затем используйте это с оператором in_. Вот пример готового продукта:

sharedFilterQuery = '''WITH    RECURSIVE
        q AS
        (
        SELECT  h.*
        FROM    "Selection"."FilterFolder" h
        join "Selection"."Filter" f
        on f."filterFolderId" = h.id
        WHERE   f.id = :filterId 
        UNION
        SELECT  hp.*
        FROM    q
        JOIN    "Selection"."FilterFolder" hp
        ON      hp.id = q."parentFolderId"
        )
SELECT  f.id
FROM    "Selection"."Filter" f
where f.id = :filterId and
(f."createdByUserId" = 1 or
exists(select 1 from q where "isShared" = TRUE LIMIT 1))
'''

inClause = text(sharedFilterQuery,bindparams=[bindparam('filterId',filterId)])

f = session.query(Filter)\
.filter(Filter.description == None)\
.filter(Filter.id.in_(inClause)).first()

Answer 2

Мне повезло с подзапросами, связывающими метод select_from с методом запроса. Нечто подобное может сработать; но не зная больше о ваших моделях данных и SQL, который вы пытаетесь сгенерировать, я не могу быть уверен.

results = session.query(sharedFilterAlias).select_from(sharedFilterQuery).all()