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.

HandlerRegistry is the central directory that connects message types to their handlers. You register handlers once at startup (or via a DI container factory), and the buses look up the correct handler at dispatch time. The registry supports two registration strategies: eager (pass a handler instance directly) and lazy (pass a zero-argument factory callable that is invoked on first use and then cached). A fluent API lets you chain register_* calls on a single registry instance. 
from hexcore.application.cqrs.registry import HandlerRegistry

​
Constructor

class HandlerRegistry:
    def __init__(self, *, allow_override: bool = False) -> None: ...
​
allow_override
bool
default:"False"
When False (the default), registering a second handler for the same command or query type raises DuplicateHandlerError. Set to True to allow handler replacement — useful for testing where you want to swap handlers without rebuilding the registry.

​
Type aliases

CommandHandlerFactory = Callable[[], ICommandHandler[Any, Any]]
QueryHandlerFactory   = Callable[[], IQueryHandler[Any, Any]]
​
CommandHandlerFactory
Callable[[], ICommandHandler]
A zero-argument callable that returns a command handler instance. Used for lazy/DI-container registration.
​
QueryHandlerFactory
Callable[[], IQueryHandler]
A zero-argument callable that returns a query handler instance. Used for lazy/DI-container registration.

​
Command handler methods

​
register_command_handler

def register_command_handler(
    self,
    command_type: type[Command],
    handler: ICommandHandler[Any, Any] | CommandHandlerFactory,
) -> HandlerRegistry
Registers a handler (or factory) for a command type. Returns self for fluent chaining.
​
command_type
type[Command]
required
The Command subclass this handler processes (e.g., CreateOrderCommand).
​
handler
ICommandHandler | CommandHandlerFactory
required
Either a handler instance (eager) or a zero-argument callable that returns a handler (lazy). Factories are invoked once on first resolution and the result is cached.
​
registry
HandlerRegistry
Returns self, enabling fluent chaining: registry.register_command_handler(...).register_command_handler(...).
Raises: DuplicateHandlerError — if command_type is already registered and allow_override=False.

​
resolve_command_handler

def resolve_command_handler(
    self, command_type: type[Command]
) -> ICommandHandler[Any, Any]
Looks up and returns the handler for command_type. If the registered entry is a factory (callable that is not an ICommandHandler), it is invoked and the resulting instance is cached, replacing the factory in the registry.
​
command_type
type[Command]
required
The command class whose handler should be resolved.
​
handler
ICommandHandler[Any, Any]
The resolved (and possibly cached) handler instance.
Raises: HandlerNotFoundError — if no handler is registered for command_type.

​
Query handler methods

​
register_query_handler

def register_query_handler(
    self,
    query_type: type[Query[Any]],
    handler: IQueryHandler[Any, Any] | QueryHandlerFactory,
) -> HandlerRegistry
Registers a handler (or factory) for a query type. Returns self for fluent chaining.
​
query_type
type[Query[Any]]
required
The Query subclass this handler processes (e.g., GetProductByIdQuery).
​
handler
IQueryHandler | QueryHandlerFactory
required
A handler instance or a zero-argument factory callable.
​
registry
HandlerRegistry
Returns self for fluent chaining.
Raises: DuplicateHandlerError — if query_type is already registered and allow_override=False.

​
resolve_query_handler

def resolve_query_handler(
    self, query_type: type[Query[Any]]
) -> IQueryHandler[Any, Any]
Looks up and returns the handler for query_type. Factory entries are invoked once and cached.
​
query_type
type[Query[Any]]
required
The query class whose handler should be resolved.
​
handler
IQueryHandler[Any, Any]
The resolved handler instance.
Raises: HandlerNotFoundError — if no handler is registered for query_type.

​
Properties

​
registered_commands

@property
def registered_commands(self) -> frozenset[type[Command]]
​
commands
frozenset[type[Command]]
An immutable set of all command types currently registered. Useful for introspection, health checks, and debugging.

​
registered_queries

@property
def registered_queries(self) -> frozenset[type[Query[Any]]]
​
queries
frozenset[type[Query[Any]]]
An immutable set of all query types currently registered.

​
Usage examples

from hexcore.application.cqrs.registry import HandlerRegistry

registry = HandlerRegistry()

(
    registry
    .register_command_handler(CreateOrderCommand, CreateOrderHandler(uow))
    .register_command_handler(CancelOrderCommand, CancelOrderHandler(uow))
    .register_query_handler(GetOrderQuery, GetOrderHandler(repo))
    .register_query_handler(ListOrdersQuery, ListOrdersHandler(repo))
)
Prefer lazy factories in production applications that use a dependency injection container. This defers construction of repositories and services until the first message is dispatched, keeping startup time predictable and enabling the container to manage object lifetimes.

Build docs developers (and LLMs) love

Get started for freeTalk to us