Events API

from sqlalchemy import event

@event.listens_for(Target, "event_name")
def receive_event(*args, **kwargs):
    # Handle event
    pass

@event.listens_for(engine, "connect")
def receive_connect(dbapi_conn, connection_record):
    print("New connection created")

@event.listens_for(engine, "checkout")
def receive_checkout(dbapi_conn, connection_record, connection_proxy):
    print("Connection checked out")

@event.listens_for(engine, "checkin")
def receive_checkin(dbapi_conn, connection_record):
    print("Connection returned")

@event.listens_for(engine, "close")
def receive_close(dbapi_conn, connection_record):
    print("Connection closed")

@event.listens_for(engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    print(f"Executing: {statement}")

@event.listens_for(engine, "after_cursor_execute")
def receive_after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    print(f"Executed: {statement}")

from sqlalchemy.orm import Session

@event.listens_for(Session, "after_transaction_create")
def receive_after_transaction_create(session, transaction):
    print("Transaction started")

@event.listens_for(Session, "after_transaction_end")
def receive_after_transaction_end(session, transaction):
    print("Transaction ended")

@event.listens_for(Session, "before_flush")
def receive_before_flush(session, flush_context, instances):
    print(f"Flushing {len(session.dirty)} dirty objects")

@event.listens_for(Session, "after_flush")
def receive_after_flush(session, flush_context):
    print("Flush complete")

@event.listens_for(Session, "after_flush_postexec")
def receive_after_flush_postexec(session, flush_context):
    print("Post-flush complete")

@event.listens_for(Session, "after_commit")
def receive_after_commit(session):
    print("Transaction committed")

@event.listens_for(Session, "after_rollback")
def receive_after_rollback(session):
    print("Transaction rolled back")

@event.listens_for(Session, "after_soft_rollback")
def receive_after_soft_rollback(session, previous_transaction):
    print("Soft rollback")

from sqlalchemy.orm import Mapper

@event.listens_for(User, "before_insert")
def receive_before_insert(mapper, connection, target):
    target.created_at = datetime.utcnow()

@event.listens_for(User, "after_insert")
def receive_after_insert(mapper, connection, target):
    print(f"Inserted user: {target.id}")

@event.listens_for(User, "before_update")
def receive_before_update(mapper, connection, target):
    target.updated_at = datetime.utcnow()

@event.listens_for(User, "after_update")
def receive_after_update(mapper, connection, target):
    print(f"Updated user: {target.id}")

@event.listens_for(User, "before_delete")
def receive_before_delete(mapper, connection, target):
    print(f"Deleting user: {target.id}")

@event.listens_for(User, "after_delete")
def receive_after_delete(mapper, connection, target):
    print(f"Deleted user: {target.id}")

@event.listens_for(User, "load")
def receive_load(target, context):
    print(f"Loaded user: {target.id}")

@event.listens_for(User, "refresh")
def receive_refresh(target, context, attrs):
    print(f"Refreshed user: {target.id}")

@event.listens_for(User, "refresh_flush")
def receive_refresh_flush(target, flush_context, attrs):
    print(f"Flush refresh user: {target.id}")

@event.listens_for(User, "init")
def receive_init(target, args, kwargs):
    print("User instance created")

@event.listens_for(User, "expire")
def receive_expire(target, attrs):
    print(f"User {target.id} expired")

from sqlalchemy.orm import attributes

@event.listens_for(User.name, "set")
def receive_set(target, value, oldvalue, initiator):
    print(f"Name changed from {oldvalue} to {value}")

@event.listens_for(User.addresses, "append")
def receive_append(target, value, initiator):
    print(f"Address added to user {target.id}")

@event.listens_for(User.addresses, "remove")
def receive_remove(target, value, initiator):
    print(f"Address removed from user {target.id}")

@event.listens_for(User, "before_insert", propagate=True)
def receive_before_insert(mapper, connection, target):
    # Applies to User and all subclasses
    pass

@event.listens_for(engine, "connect", raw=True)
def receive_connect(dbapi_conn):
    # dbapi_conn is raw driver connection
    pass

@event.listens_for(Session, "before_flush", retval=True)
def receive_before_flush(session, flush_context, instances):
    # Can prevent flush by returning False
    return True

from datetime import datetime

@event.listens_for(User, "before_insert")
def set_created_at(mapper, connection, target):
    target.created_at = datetime.utcnow()

@event.listens_for(User, "before_update")
def set_updated_at(mapper, connection, target):
    target.updated_at = datetime.utcnow()

@event.listens_for(Session, "after_flush")
def receive_after_flush(session, flush_context):
    for obj in session.new:
        audit_log("INSERT", obj)
    for obj in session.dirty:
        audit_log("UPDATE", obj)
    for obj in session.deleted:
        audit_log("DELETE", obj)

import logging

logger = logging.getLogger("sqlalchemy.engine")

@event.listens_for(engine, "before_cursor_execute")
def log_query(conn, cursor, statement, parameters, context, executemany):
    logger.info(f"SQL: {statement}")
    logger.info(f"Params: {parameters}")

@event.listens_for(engine, "connect")
def set_connection_options(dbapi_conn, connection_record):
    cursor = dbapi_conn.cursor()
    cursor.execute("SET TIME ZONE 'UTC'")
    cursor.close()