Registry API

This module contains the event handler registry and signals used by Django Dispatch.

EventHandlerRegistry

class django_broadcaster.registry.EventHandlerRegistry

Bases: object

Registry for event handlers that can process events locally

__init__()

register(event_type: str | None = None)

Decorator to register an event handler

Parameters:

event_type – Specific event type to handle, None for all events

handle_event(event: OutboxEvent) → bool

Handle an event by calling registered handlers

Parameters:

event – The event to handle

Returns:

True if all handlers succeeded, False otherwise

Return type:

bool

get_handlers(event_type: str | None = None) → List[Callable]

Get registered handlers for an event type

clear_handlers(event_type: str | None = None)

Clear handlers for an event type or all handlers

clear_handlers(event_type: str | None = None)

Clear handlers for an event type or all handlers

get_handlers(event_type: str | None = None) → List[Callable]

Get registered handlers for an event type

handle_event(event: OutboxEvent) → bool

Handle an event by calling registered handlers

Parameters:

event – The event to handle

Returns:

True if all handlers succeeded, False otherwise

Return type:

bool

register(event_type: str | None = None)

Decorator to register an event handler

Parameters:

event_type – Specific event type to handle, None for all events

Singleton Instance

django_broadcaster.registry.event_registry

A singleton instance of EventHandlerRegistry that is created when the module is imported. This is the recommended way to access the registry functionality.

Example usage:

from django_broadcaster.registry import event_registry

@event_registry.register('user.created')
def handle_user_created(event):
    print(f"User created: {event.data}")

Signals

django_broadcaster.registry.event_created

A Django signal that is dispatched when an event is created.

Signal arguments:

• sender: The sender of the signal (OutboxEvent class)

• event: The OutboxEvent instance that was created

Example usage:

from django.dispatch import receiver
from django_broadcaster.registry import event_created

@receiver(event_created)
def handle_event_created(sender, event, **kwargs):
    print(f"Event created: {event.event_type}")

django_broadcaster.registry.event_published

A Django signal that is dispatched when an event is published.

Signal arguments:

• sender: The sender of the signal (OutboxEvent class)

• event: The OutboxEvent instance that was published

Example usage:

from django.dispatch import receiver
from django_broadcaster.registry import event_published

@receiver(event_published)
def handle_event_published(sender, event, **kwargs):
    print(f"Event published: {event.event_type}")