Dominik George Dominik George - 2 months ago 6x
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 (

, etc.) is executed.

I have thought about the following approaches:

1. Subclass the Query class

I could subclass
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) ;).


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:
        # do something with query.statement

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