Event Bus

event_bus

Event Bus infrastructure.

This module provides the EventBusPort port and its in-memory implementation.

EventBusPort

Bases: ABC

EventBusPort port.

The EventBusPort is responsible for publishing events and sending commands
to their respective handlers. It follows the publish-subscribe pattern
for events and the command pattern for commands.

Source code in src/forging_blocks/infrastructure/event_bus.py

class EventBusPort(ABC):
    """
    EventBusPort port.

    The EventBusPort is responsible for publishing events and sending commands
    to their respective handlers. It follows the publish-subscribe pattern
    for events and the command pattern for commands.
    """

    @abstractmethod
    async def publish(self, event: Event[Any]) -> None:
        """
        Publish an event to all registered handlers.

        Args:
            event: The event to publish.
        """
        pass

    @abstractmethod
    async def send(self, command: Command[Any]) -> None:
        """
        Send a command to its handler.

        Args:
            command: The command to send.

        Raises:
            NoHandlerError: If no handler is registered for the command type.
        """
        pass

    @abstractmethod
    def subscribe(self, event_type: Type[E], handler: EventHandler) -> None:
        """
        Subscribe a handler to an event type.

        Args:
            event_type: The type of event to subscribe to.
            handler: The handler function to call when the event is published.
        """
        pass

    @abstractmethod
    def register_command_handler(self, command_type: Type[C], handler: CommandHandler) -> None:
        """
        Register a handler for a command type.

        Args:
            command_type: The type of command to handle.
            handler: The handler function to call when the command is sent.
        """
        pass

publish(event: Event[Any]) -> None abstractmethod async

Publish an event to all registered handlers.

Parameters:

Name	Type	Description	Default
event	Event[Any]	The event to publish.	required

Source code in src/forging_blocks/infrastructure/event_bus.py

@abstractmethod
async def publish(self, event: Event[Any]) -> None:
    """
    Publish an event to all registered handlers.

    Args:
        event: The event to publish.
    """
    pass

send(command: Command[Any]) -> None abstractmethod async

Send a command to its handler.

Parameters:

Name	Type	Description	Default
command	Command[Any]	The command to send.	required

Raises:

Type	Description
NoHandlerError	If no handler is registered for the command type.

Source code in src/forging_blocks/infrastructure/event_bus.py

@abstractmethod
async def send(self, command: Command[Any]) -> None:
    """
    Send a command to its handler.

    Args:
        command: The command to send.

    Raises:
        NoHandlerError: If no handler is registered for the command type.
    """
    pass

subscribe(event_type: Type[E], handler: EventHandler) -> None abstractmethod

Subscribe a handler to an event type.

Parameters:

Name	Type	Description	Default
event_type	Type[E]	The type of event to subscribe to.	required
handler	EventHandler	The handler function to call when the event is published.	required

Source code in src/forging_blocks/infrastructure/event_bus.py

@abstractmethod
def subscribe(self, event_type: Type[E], handler: EventHandler) -> None:
    """
    Subscribe a handler to an event type.

    Args:
        event_type: The type of event to subscribe to.
        handler: The handler function to call when the event is published.
    """
    pass

register_command_handler(command_type: Type[C], handler: CommandHandler) -> None abstractmethod

Register a handler for a command type.

Parameters:

Name	Type	Description	Default
command_type	Type[C]	The type of command to handle.	required
handler	CommandHandler	The handler function to call when the command is sent.	required

Source code in src/forging_blocks/infrastructure/event_bus.py

@abstractmethod
def register_command_handler(self, command_type: Type[C], handler: CommandHandler) -> None:
    """
    Register a handler for a command type.

    Args:
        command_type: The type of command to handle.
        handler: The handler function to call when the command is sent.
    """
    pass

NoHandlerError

Bases: Error[dict[str, object]]

Raised when no handler is registered for a command type.

Source code in src/forging_blocks/infrastructure/event_bus.py

class NoHandlerError(Error[dict[str, object]]):
    """
    Raised when no handler is registered for a command type.
    """

    def __init__(self, message: str) -> None:
        super().__init__(ErrorMessage(message))