Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt

Use this file to discover all available pages before exploring further.

Buses are the message-passing layer of CQRS. They decouple the sender from the handler: callers dispatch a command or ask a query without knowing which class handles it or where it runs. HexCore provides three abstract ports — AbstractCommandBus, AbstractQueryBus, and AbstractEventBus — plus in-memory implementations (InMemoryCommandBus, InMemoryQueryBus, InMemoryEventBus) that resolve handlers from a HandlerRegistry, run them through a MiddlewarePipeline, and support Smart Routing for background execution.

​
CQRS exceptions

Both abstract and concrete buses raise typed exceptions from hexcore.domain.cqrs.exceptions:

​
HandlerNotFoundError

from hexcore.domain.cqrs.exceptions import HandlerNotFoundError

class HandlerNotFoundError(CQRSError):
    def __init__(self, message_type: type) -> None: ...
Raised by any bus when no handler is registered for the dispatched command, query, or event type.
​
message_type
type
required
The command, query, or event class that had no registered handler.

​
DuplicateHandlerError

from hexcore.domain.cqrs.exceptions import DuplicateHandlerError

class DuplicateHandlerError(CQRSError):
    def __init__(self, message_type: type) -> None: ...
Raised by HandlerRegistry when attempting to register a second handler for a type that already has one and allow_override=False.
​
message_type
type
required
The command or query class for which a duplicate registration was attempted.

​
Abstract contracts

​
AbstractCommandBus

from hexcore.domain.cqrs.buses import AbstractCommandBus

class AbstractCommandBus(abc.ABC):
    @abstractmethod
    async def dispatch(self, command: Command) -> Any: ...
The port for sending commands. Implementations resolve the handler, run the middleware pipeline, and return the handler’s result.

​
dispatch

@abstractmethod
async def dispatch(self, command: Command) -> Any
​
command
Command
required
The command instance to dispatch.
​
result
Any
The value returned by the command handler. None for fire-and-forget commands.
Raises: HandlerNotFoundError — if no handler is registered for the command’s type.

​
AbstractQueryBus

from hexcore.domain.cqrs.buses import AbstractQueryBus

class AbstractQueryBus(abc.ABC):
    @abstractmethod
    async def ask(self, query: Query[TResult]) -> TResult: ...
The port for issuing queries. Returns a strongly-typed result matching the query’s TResult parameter.

​
ask

@abstractmethod
async def ask(self, query: Query[TResult]) -> TResult
​
query
Query[TResult]
required
The query instance to resolve.
​
result
TResult
The data returned by the query handler, typed to match the query’s generic parameter.
Raises: HandlerNotFoundError — if no handler is registered for the query’s type.

​
AbstractEventBus

from hexcore.domain.cqrs.buses import AbstractEventBus

class AbstractEventBus(abc.ABC):
    @abstractmethod
    async def publish(self, event: DomainEvent) -> None: ...

    @abstractmethod
    def subscribe(
        self,
        event_type: type[DomainEvent],
        handler: Callable[[DomainEvent], Awaitable[None]],
    ) -> None: ...
The port for domain event publishing and subscription.

​
publish

@abstractmethod
async def publish(self, event: DomainEvent) -> None
​
event
DomainEvent
required
The domain event to publish to all registered subscribers.

​
subscribe

@abstractmethod
def subscribe(
    self,
    event_type: type[DomainEvent],
    handler: Callable[[DomainEvent], Awaitable[None]],
) -> None
​
event_type
type[DomainEvent]
required
The event class to subscribe to.
​
handler
Callable[[DomainEvent], Awaitable[None]]
required
An async callable invoked whenever an event of event_type is published.

​
Backward-compatibility aliases

ICommandBus = AbstractCommandBus
IQueryBus   = AbstractQueryBus
IEventBus   = AbstractEventBus
These aliases are importable from hexcore.domain.cqrs.buses and are fully equivalent.

​
In-memory implementations

​
InMemoryCommandBus

from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
A synchronous in-memory command bus with Smart Routing support. Commands decorated with @background_command are automatically enqueued to a task backend instead of being handled locally, provided enqueuer and serializer are configured.

​
Constructor

class InMemoryCommandBus(ICommandBus):
    def __init__(
        self,
        registry: HandlerRegistry,
        pipeline: MiddlewarePipeline | None = None,
        enqueuer: ITaskEnqueuer | None = None,
        serializer: ISerializer | None = None,
    ) -> None: ...
​
registry
HandlerRegistry
required
The handler registry used to resolve command handlers.
​
pipeline
MiddlewarePipeline | None
default:"MiddlewarePipeline()"
Optional middleware pipeline. An empty pipeline is used when not provided.
​
enqueuer
ITaskEnqueuer | None
default:"None"
Task queue adapter for Smart Routing. Required if any registered command uses @background_command.
​
serializer
ISerializer | None
default:"None"
Serializer for converting command instances to task payloads. Required alongside enqueuer.

​
dispatch

async def dispatch(self, command: Command) -> Any
Smart Routing logic:
  1. Checks type(command).__cqrs_background__. If True and enqueuer/serializer are configured, the command payload is serialised and sent to the task queue (using the queue name from __cqrs_queue__, defaulting to "default"). Returns None immediately.
  2. If the command is not marked for background execution, resolves the handler from the registry and executes it through the middleware pipeline synchronously.
Dispatching a @background_command without configuring enqueuer and serializer raises a RuntimeError at runtime. Ensure both are provided when using background commands.

​
InMemoryQueryBus

from hexcore.application.cqrs.in_memory_buses import InMemoryQueryBus
A synchronous in-memory query bus. Queries are never routed to background workers — reads are always executed locally and return results immediately.

​
Constructor

class InMemoryQueryBus(IQueryBus):
    def __init__(
        self,
        registry: HandlerRegistry,
        pipeline: MiddlewarePipeline | None = None,
    ) -> None: ...
​
registry
HandlerRegistry
required
The handler registry used to resolve query handlers.
​
pipeline
MiddlewarePipeline | None
default:"MiddlewarePipeline()"
Optional middleware pipeline applied before the handler is called.

​
ask

async def ask(self, query: Query[Any]) -> Any
Resolves the handler for type(query) from the registry and executes it through the middleware pipeline, returning the handler’s result.

​
InMemoryEventBus

from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
An in-memory event bus that supports multiple subscribers per event type and Smart Routing for individual handlers decorated with @background_handler. Each handler is evaluated independently: some may run locally while others are routed to a task queue.

​
Constructor

class InMemoryEventBus(IEventBus):
    def __init__(
        self,
        pipeline: MiddlewarePipeline | None = None,
        enqueuer: ITaskEnqueuer | None = None,
        serializer: ISerializer | None = None,
    ) -> None: ...
​
pipeline
MiddlewarePipeline | None
default:"MiddlewarePipeline()"
Middleware pipeline applied to every published event before handlers are invoked.
​
enqueuer
ITaskEnqueuer | None
default:"None"
Task queue adapter. Required for handlers decorated with @background_handler.
​
serializer
ISerializer | None
default:"None"
Serializer for converting event instances to task payloads. Required alongside enqueuer.

​
subscribe

def subscribe(
    self,
    event_type: type[DomainEvent],
    handler: Callable[[DomainEvent], Awaitable[None]],
) -> None
Appends handler to the subscriber list for event_type. Multiple handlers may be registered for the same event type.
​
event_type
type[DomainEvent]
required
The event class to subscribe to.
​
handler
Callable[[DomainEvent], Awaitable[None]]
required
The async event handler callable.

​
publish

async def publish(self, event: DomainEvent) -> None
Iterates over all subscribers registered for type(event). For each subscriber:
  • If the handler has __cqrs_background_handler__ = True and enqueuer/serializer are set, the event payload is serialised and enqueued using enqueuer.enqueue_handler(handler_name, payload, queue=queue_name).
  • Otherwise, the handler is called directly through the middleware pipeline.

​
Wire-up example

from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.application.cqrs.pipeline import MiddlewarePipeline
from hexcore.application.cqrs.in_memory_buses import (
    InMemoryCommandBus,
    InMemoryQueryBus,
    InMemoryEventBus,
)

registry = HandlerRegistry()
registry.register_command_handler(CreateOrderCommand, CreateOrderHandler(uow))
registry.register_query_handler(GetOrderQuery, GetOrderHandler(repo))

command_bus = InMemoryCommandBus(registry=registry)
query_bus   = InMemoryQueryBus(registry=registry)
event_bus   = InMemoryEventBus()

event_bus.subscribe(OrderCreatedEvent, send_confirmation_email)

Build docs developers (and LLMs) love

Get started for freeTalk to us