Publishers API

This module contains the publisher classes used to publish events to the outbox.

OutboxPublisher

class django_broadcaster.publishers.OutboxPublisher[source]

Bases: object

Main publisher class for the outbox pattern

__init__()[source]
_load_backends()[source]

Load configured publisher backends

publish_event(event_type: str, source: str, data: Any | None = None, subject: str = '', backend: str | None = None, scheduled_at: datetime | None = None, max_retries: int = 3, publisher_config: Dict[str, Any] | None = None) → OutboxEvent[source]

Publish an event to the outbox

Parameters:

  • event_type – Type of the event

  • source – Source of the event

  • data – Event payload data

  • subject – Event subject (optional)

  • backend – Publisher backend to use (optional)

  • scheduled_at – When to publish the event (default: now)

  • max_retries – Maximum retry attempts

  • publisher_config – Backend-specific configuration

Returns:

The created event

Return type:

OutboxEvent

publish_cloud_event(cloud_event: CloudEvent, backend: str | None = None, scheduled_at: datetime | None = None, max_retries: int = 3, publisher_config: Dict[str, Any] | None = None) → OutboxEvent[source]

Publish a CloudEvent to the outbox

process_pending_events(batch_size: int = 100) → int[source]

Process pending events from the outbox

Parameters:

batch_size – Number of events to process in one batch

Returns:

Number of events processed

Return type:

int

_process_single_event(event: OutboxEvent) → bool[source]

Process a single event

Parameters:

event – The event to process

Returns:

True if successfully processed, False otherwise

Return type:

bool

_get_default_backend() → str | None[source]

Get the default backend name

get_backend_health() → Dict[str, bool][source]

Get health status of all backends

get_backend_health() → Dict[str, bool][source]

Get health status of all backends

process_pending_events(batch_size: int = 100) → int[source]

Process pending events from the outbox

Parameters:

batch_size – Number of events to process in one batch

Returns:

Number of events processed

Return type:

int

publish_cloud_event(cloud_event: CloudEvent, backend: str | None = None, scheduled_at: datetime | None = None, max_retries: int = 3, publisher_config: Dict[str, Any] | None = None) → OutboxEvent[source]

Publish a CloudEvent to the outbox

publish_event(event_type: str, source: str, data: Any | None = None, subject: str = '', backend: str | None = None, scheduled_at: datetime | None = None, max_retries: int = 3, publisher_config: Dict[str, Any] | None = None) → OutboxEvent[source]

Publish an event to the outbox

Parameters:

  • event_type – Type of the event

  • source – Source of the event

  • data – Event payload data

  • subject – Event subject (optional)

  • backend – Publisher backend to use (optional)

  • scheduled_at – When to publish the event (default: now)

  • max_retries – Maximum retry attempts

  • publisher_config – Backend-specific configuration

Returns:

The created event

Return type:

OutboxEvent

Singleton Instance

django_broadcaster.publishers.publisher

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

Example usage:

from django_broadcaster.publishers import publisher

event = publisher.publish_event(
    event_type='user.created',
    source='user-service',
    data={'user_id': 123}
)