Skip to content

Events

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

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)

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