Skip to content

In Memory Event Bus Base

in_memory_event_bus_base

In-memory implementation of the EventBusBase.

Dispatches events to multiple registered handlers (fan-out) and
commands to a single registered handler. Handlers are looked up
by the exact type of the message.

InMemoryEventBusBase

Bases: EventBusBase[EventPayloadType, CommandPayloadType, HandlerType]

In-memory event bus with separate event/command dispatch.

Attributes:

Name Type Description
_event_handlers dict[type[Event[EventPayloadType]], list[_Handler[Event[EventPayloadType]]]]

Per-event-type list of handlers.
_command_handlers dict[type[Command[CommandPayloadType]], _Handler[Command[CommandPayloadType]]]

Per-command-type single handler.
Example 
class StubEvent[T]:
    def __init__(self) -> None:
        pass


class StubCommand[T]:
    def __init__(self) -> None:
        pass


class OrderCompleted(StubEvent[dict[str, object]]):
    def __init__(self, order_id: str) -> None:
        self.order_id = order_id


class CreateOrder(StubCommand[dict[str, object]]):
    def __init__(self, customer_id: str) -> None:
        self.customer_id = customer_id


class OrderCompletedHandler:
    async def handle(self, event: OrderCompleted) -> None:
        print(f"Order completed: {event.order_id}")


class CreateOrderHandler:
    async def handle(self, command: CreateOrder) -> None:
        print(f"Creating order for: {command.customer_id}")


bus = InMemoryEventBusBase[dict[str, object], dict[str, object], object]()
bus.register_handler(OrderCompleted, OrderCompletedHandler())
bus.register_handler(CreateOrder, CreateOrderHandler())
await bus.publish(OrderCompleted(order_id="abc-123"))
await bus.send(CreateOrder(customer_id="cust-42"))
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus_base.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
class InMemoryEventBusBase[EventPayloadType, CommandPayloadType, HandlerType](
    EventBusBase[EventPayloadType, CommandPayloadType, HandlerType]
):
    """In-memory event bus with separate event/command dispatch.

    Attributes:
        _event_handlers: Per-event-type list of handlers.
        _command_handlers: Per-command-type single handler.

    Example:
        ```python
        class StubEvent[T]:
            def __init__(self) -> None:
                pass


        class StubCommand[T]:
            def __init__(self) -> None:
                pass


        class OrderCompleted(StubEvent[dict[str, object]]):
            def __init__(self, order_id: str) -> None:
                self.order_id = order_id


        class CreateOrder(StubCommand[dict[str, object]]):
            def __init__(self, customer_id: str) -> None:
                self.customer_id = customer_id


        class OrderCompletedHandler:
            async def handle(self, event: OrderCompleted) -> None:
                print(f"Order completed: {event.order_id}")


        class CreateOrderHandler:
            async def handle(self, command: CreateOrder) -> None:
                print(f"Creating order for: {command.customer_id}")


        bus = InMemoryEventBusBase[dict[str, object], dict[str, object], object]()
        bus.register_handler(OrderCompleted, OrderCompletedHandler())
        bus.register_handler(CreateOrder, CreateOrderHandler())
        await bus.publish(OrderCompleted(order_id="abc-123"))
        await bus.send(CreateOrder(customer_id="cust-42"))
        ```
    """

    __slots__ = ("_command_handlers", "_event_handlers")

    def __init__(self) -> None:
        self._event_handlers: dict[
            type[Event[EventPayloadType]], list[_Handler[Event[EventPayloadType]]]
        ] = {}
        self._command_handlers: dict[
            type[Command[CommandPayloadType]], _Handler[Command[CommandPayloadType]]
        ] = {}

    def register_handler(
        self,
        message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
        handler: HandlerType,
    ) -> None:
        """Register a handler for a message type.

        For event types, multiple handlers can be registered (fan-out).
        For command types, only one handler is allowed per type.

        Args:
            message_type: The message class to handle.
            handler: A handler instance.

        """
        if issubclass(message_type, Event):
            self._event_handlers.setdefault(message_type, []).append(
                cast(_Handler[Event[EventPayloadType]], handler)
            )
        else:
            self._command_handlers[message_type] = cast(
                _Handler[Command[CommandPayloadType]], handler
            )

    async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
        """Publish an event to all registered handlers.

        Args:
            event: The domain event.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
            handler raises.

        """
        handlers = self._event_handlers.get(type(event), [])
        for handler in handlers:
            try:
                await handler.handle(event)
            except Exception as exc:
                return Err(EventBusError(str(exc)))
        return Ok(None)

    async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
        """Send a command to its registered handler.

        Args:
            command: The command.

        Returns:
            ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
            handler raises or no handler is registered.

        """
        handler = self._command_handlers.get(type(command))
        if handler is None:
            return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
        try:
            await handler.handle(command)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
        return Ok(None)

register_handler(message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]], handler: HandlerType) -> None

Register a handler for a message type.

For event types, multiple handlers can be registered (fan-out).
For command types, only one handler is allowed per type.

Parameters:

Name Type Description Default
message_type type[Event[EventPayloadType]] | type[Command[CommandPayloadType]]

The message class to handle.

 required
handler HandlerType

A handler instance.

 required
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus_base.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def register_handler(
    self,
    message_type: type[Event[EventPayloadType]] | type[Command[CommandPayloadType]],
    handler: HandlerType,
) -> None:
    """Register a handler for a message type.

    For event types, multiple handlers can be registered (fan-out).
    For command types, only one handler is allowed per type.

    Args:
        message_type: The message class to handle.
        handler: A handler instance.

    """
    if issubclass(message_type, Event):
        self._event_handlers.setdefault(message_type, []).append(
            cast(_Handler[Event[EventPayloadType]], handler)
        )
    else:
        self._command_handlers[message_type] = cast(
            _Handler[Command[CommandPayloadType]], handler
        )

publish(event: Event[EventPayloadType]) -> Result[None, EventBusError] async

Publish an event to all registered handlers.

Parameters:

Name Type Description Default
event Event[EventPayloadType]

The domain event.

 required

Returns:

Type Description
Result[None, EventBusError]

Ok(None) on success, or Err(EventBusError) if any
Result[None, EventBusError]

handler raises.
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus_base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def publish(self, event: Event[EventPayloadType]) -> Result[None, EventBusError]:
    """Publish an event to all registered handlers.

    Args:
        event: The domain event.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if any
        handler raises.

    """
    handlers = self._event_handlers.get(type(event), [])
    for handler in handlers:
        try:
            await handler.handle(event)
        except Exception as exc:
            return Err(EventBusError(str(exc)))
    return Ok(None)

send(command: Command[CommandPayloadType]) -> Result[None, EventBusError] async

Send a command to its registered handler.

Parameters:

Name Type Description Default
command Command[CommandPayloadType]

The command.

 required

Returns:

Type Description
Result[None, EventBusError]

Ok(None) on success, or Err(EventBusError) if the
Result[None, EventBusError]

handler raises or no handler is registered.
Source code in src/forging_blocks/infrastructure/event_buses/in_memory_event_bus_base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
async def send(self, command: Command[CommandPayloadType]) -> Result[None, EventBusError]:
    """Send a command to its registered handler.

    Args:
        command: The command.

    Returns:
        ``Ok(None)`` on success, or ``Err(EventBusError)`` if the
        handler raises or no handler is registered.

    """
    handler = self._command_handlers.get(type(command))
    if handler is None:
        return Err(EventBusError(f"No handler registered for {type(command).__name__}"))
    try:
        await handler.handle(command)
    except Exception as exc:
        return Err(EventBusError(str(exc)))
    return Ok(None)