Skip to content

Events

Events let applications register sync or async handlers around ORM lifecycle operations.

Common transaction and unit-of-work events:

Event When it fires
before_flush, after_flush Around Session.flush().
before_begin, after_begin Around transaction begin.
before_commit, after_commit Around transaction commit.
before_rollback, after_rollback Around transaction rollback.
before_savepoint, after_savepoint Around savepoint creation.
before_release_savepoint, after_release_savepoint Around successful savepoint release.
before_rollback_to_savepoint, after_rollback_to_savepoint Around savepoint rollback.

after_* transaction events include duration_ms and error; error is None when the operation succeeds.

ormdantic.events.EventHandler module-attribute 

EventHandler = Callable[..., Any | Awaitable[Any]]

ormdantic.events.EventRegistry 

EventRegistry()

Store and dispatch sync or async event handlers.

Create an empty event registry.

Source code in ormdantic/events.py 
def __init__(self) -> None:
    """Create an empty event registry."""
    self._handlers: dict[str, list[EventHandler]] = defaultdict(list)

on 

on(event, handler)

Register a handler for an event and return the handler.

Source code in ormdantic/events.py 
def on(self, event: str, handler: EventHandler) -> EventHandler:
    """Register a handler for an event and return the handler."""
    self._handlers[event].append(handler)
    return handler

off 

off(event, handler)

Remove a previously registered event handler.

Source code in ormdantic/events.py 
def off(self, event: str, handler: EventHandler) -> None:
    """Remove a previously registered event handler."""
    if handler in self._handlers.get(event, []):
        self._handlers[event].remove(handler)

clear 

clear(event=None)

Clear handlers for one event or for all events.

Source code in ormdantic/events.py 
def clear(self, event: str | None = None) -> None:
    """Clear handlers for one event or for all events."""
    if event is None:
        self._handlers.clear()
    else:
        self._handlers.pop(event, None)

has_handlers 

has_handlers(event)

Return whether handlers are registered for an event.

Source code in ormdantic/events.py 
def has_handlers(self, event: str) -> bool:
    """Return whether handlers are registered for an event."""
    return bool(self._handlers.get(event))

dispatch async 

dispatch(event, **payload)

Dispatch an event payload to all registered handlers.

Source code in ormdantic/events.py 
async def dispatch(self, event: str, **payload: Any) -> None:
    """Dispatch an event payload to all registered handlers."""
    for handler in self._handlers.get(event, []):
        result = handler(**payload)
        if inspect.isawaitable(result):
            await result