Listening to sql events with SQLAlchemy

Here is an example of how to listen to sqlalchemy events:

from datetime import UTC, datetime
from typing import TYPE_CHECKING

from models import User

if TYPE_CHECKING:
    from sqlalchemy.engine import Connection
    from sqlalchemy.orm.mapper import Mapper


@event.listens_for(User, "after_insert")
@event.listens_for(User, "after_update")
def update_date(mapper: "Mapper", connection: "Connection", target: User):
        connection.execute(
            User.__table__.update()
            .where(User.__table__.c.id == target.id)
            .values(update_time=datetime.now(UTC))
        )

Here is a list of events that can be listened to.

Gotchas

The following events are only triggered when a flush is performed:

  • before_insert
  • after_insert
  • before_update
  • after_update
  • before_delete
  • after_delete

From the documentation:

It is important to note that these events apply only to the session flush operation, and not to the ORM-level INSERT/UPDATE/DELETE functionality described at ORM-Enabled INSERT, UPDATE, and DELETE statements. To intercept ORM-level DML, use the SessionEvents.do_orm_execute() event.

So, the event will always be triggered by:

  • Session.commit
  • Session.begin_nested
  • Session.prepare

It will be triggered, if autoflush is enabled, by:

  • Session.execute
  • Invoking a query
  • Session.merge
  • Refreshing objects
  • ORM Lazy loads

It will not be triggered by the following methods, imported from sqlalchemy, directly:

  • insert
  • update
  • delete
  • bindparam

So the event listener will be triggered by:

from sqlalchemy.orm import Session

from models import User

session = Session()
new_user = User(name="John", email="john@doe.com")
session.add(new_user)
session.commit()

However, this will not trigger the event listener:

from sqlalchemy import insert

from models import User

stmt = (
    insert(User).
    values(name="John", email="john@doe.com")
)

Bonus: printing compiled statements

from sqlalchemy import insert
from sqlalchemy.dialects import postgresql

from models import User

statement = (
    insert(User).
    values(name="John", email="john@doe.com")
)
print(statement.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))