SQLAlchemy - Combine Textual query with a filter

问题:

I'm using SA 0.6.6, Python 2.66 and Postgres 8.3.

I have certain queries which require somewhat complex security check that can be handled with a WITH RECURSIVE query. What I'm trying to do is combine a textual query with a query object so I can apply filters as necessary.

My original thought was was to create my text query as a subquery and then combine that with the user's query and filters. Unfortunately, this isn't working.

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

This results in this error:

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

Is there anyway to combine a textual query with SQLAlchemy's query object?

回答1:

After a time going by without an answer I posted to the SA Google Group, where Michael Bayer himself set me in the right direction.

The answer is to turn my text query into an SA text clause. Then use that with in_ operator. Here's an example of the finished product:

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()

回答2:

I've had luck with subqueries chaining the select_from method to query method. Something like this might work; but not knowing more about your data models and the SQL you're trying to generate I cannot be certain.

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