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