Backends API

This module contains the backend classes used to publish events to external systems.

PublisherBackend

class django_broadcaster.backends.PublisherBackend

Bases: ABC

Abstract base class for publisher backends

abstractmethod publish(event: OutboxEvent) → bool

Publish an event to the backend

Parameters:

event – The OutboxEvent to publish

Returns:

True if successful, False otherwise

Return type:

bool

Raises:

Exception – If publishing fails

abstractmethod health_check() → bool

Check if the backend is healthy and available

Returns:

True if healthy, False otherwise

Return type:

bool

get_name() → str

Get the backend name

get_name() → str

Get the backend name

abstractmethod health_check() → bool

Check if the backend is healthy and available

Returns:

True if healthy, False otherwise

Return type:

bool

abstractmethod publish(event: OutboxEvent) → bool

Publish an event to the backend

Parameters:

event – The OutboxEvent to publish

Returns:

True if successful, False otherwise

Return type:

bool

Raises:

Exception – If publishing fails

RedisStreamBackend

class django_broadcaster.backends.RedisStreamBackend(config: Dict[str, Any])

Bases: PublisherBackend

Redis Stream publisher backend

__init__(config: Dict[str, Any])

publish(event: OutboxEvent) → bool

Publish event to Redis Stream

health_check() → bool

Check Redis connection health

get_name() → str

Get the backend name

get_name() → str

Get the backend name

health_check() → bool

Check Redis connection health

publish(event: OutboxEvent) → bool

Publish event to Redis Stream

Creating Custom Backends

To create a custom backend, you need to subclass PublisherBackend and implement the required methods:

from django_broadcaster.backends import PublisherBackend
from django_broadcaster.models import OutboxEvent

class MyCustomBackend(PublisherBackend):
    """
    Custom publisher backend implementation
    """

    def __init__(self, config):
        self.config = config
        # Initialize your backend with the provided configuration

    def publish(self, event: OutboxEvent) -> bool:
        """
        Publish an event to your custom backend
        """
        try:
            # Convert to CloudEvent format
            cloud_event = event.to_cloud_event()

            # Implement your publishing logic here
            # ...

            return True
        except Exception as e:
            # Log the error and re-raise
            raise

    def health_check(self) -> bool:
        """
        Check if the backend is healthy and available
        """
        try:
            # Implement your health check logic here
            # ...
            return True
        except Exception:
            return False