Dominik George Dominik George - 3 months ago 9
Python Question

Intercept all queries on a model in SQLAlchemy

I need to intercept all queries that concern a model in SQLAlchemy, in a way that I can inspect it at the point where any of the query methods (

all()
,
one()
,
scalar()
, etc.) is executed.

I have thought about the following approaches:

1. Subclass the Query class



I could subclass
sqlalchemy.orm.Query
and override the execution code, starting basically from something like this.

However, I am writing a library that can be used in other SQLAlchemy applications, and thus the creation of the declarative base, let alone engines and sessions, is outside my scope.

Maybe I have missed something and it is possible to override the Query class for my models without knowledge of the session?

2. Use the before_execute Core Event



I have also thought of hooking into execution with the before_execute event.

The problem is thatit is bound to an engine (see above). Also, I need to modify objects in the session, and I got the impression that I do not have access to a session from within this event.




What I want to be able to do is something like:


  1. session.query(MyModel).filter_by(foo="bar").all()
    is executed.

  2. Intercept that query and do something like storing the query in a log table within the same database (not literally that, but a set of different things that basically need the exact same functionality as this example operation)

  3. Let the query execute like normal.



What I am trying to do in the end is inject items from another data store into the SQLAlchemy database on-the-fly upon querying. While this seems stupid - trust me, it might be less stupid than it sounds (or even more stupid) ;).

Answer

The before_compile query event might be useful for you.

from weakref import WeakSet
from sqlalchemy import event
from sqlalchemy.orm import Query

visited_queries = WeakSet()

@event.listens_for(Query, 'before_compile')
def log_query(query):
    # You can get the session
    session = query.session

    # Prevent recursion if you want to compile the query to log it!
    if query not in visited_queries:
        visited_queries.add(query)
        # do something with query.statement

You can look at query.column_descriptions to see if your model is being queried.

Comments