CODEWITHSUNDEEP
pythonsqlalchemy
user3035850
user3035850

Reputation:

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) ;).

Upvotes: 3

Views: 2505

Answers (1)

RazerM
RazerM

Reputation: 5482

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.

Upvotes: 1

Related Questions