Route CQRS Messages to Background Workers Automatically

Smart Routing lets you move commands, event handlers, and generic tasks to background worker queues by adding a single decorator — no manual queue calls, no serialization boilerplate, no conditional dispatch logic scattered through your codebase. HexCore’s buses detect the routing decorators at dispatch time and automatically delegate to the configured ITaskEnqueuer, keeping your application code clean and queue-agnostic.

The three routing decorators

All three decorators live in hexcore.domain.cqrs.decorators and work by setting special attributes on the class or function they decorate.

`@background_command(queue="...")`

Marks a Command subclass so that InMemoryCommandBus always routes it to a background queue instead of executing the handler in-process.

# hexcore/domain/cqrs/decorators.py
def background_command(queue: str = "default"):
    def wrapper(cls):
        cls.__cqrs_background__ = True
        cls.__cqrs_queue__ = queue
        return cls
    return wrapper

When the bus detects __cqrs_background__ = True, it calls enqueuer.enqueue_command(command_name, payload, queue) and returns None immediately — the handler runs asynchronously in a worker process.

from hexcore.domain.cqrs.commands import Command
from hexcore.domain.cqrs.decorators import background_command

@background_command(queue="emails")
class SendWelcomeEmailCommand(Command):
    user_id: str
    email: str

`@background_handler(queue="...")`

Marks an event handler function so that InMemoryEventBus (and RedisEventBus) routes that specific handler to a background queue. Other handlers subscribed to the same event are unaffected.

# hexcore/domain/cqrs/decorators.py
def background_handler(queue: str = "default"):
    def wrapper(func):
        func.__cqrs_background_handler__ = True
        func.__cqrs_queue__ = queue
        func.__cqrs_handler_name__ = f"{func.__module__}.{func.__qualname__}"
        return func
    return wrapper

from hexcore.domain.cqrs.decorators import background_handler
from app.events import UserRegisteredEvent

@background_handler(queue="analytics")
async def on_user_registered_analytics(event: UserRegisteredEvent) -> None:
    await track_signup_event(event.user_id)

# This handler runs in-process (no decorator)
async def on_user_registered_welcome(event: UserRegisteredEvent) -> None:
    await send_welcome_push_notification(event.user_id)

`@background_task(queue="...")`

Marks a generic async function as a background task, independent of CQRS. Register it with the worker and enqueue it via ITaskEnqueuer.enqueue_task().

# hexcore/domain/cqrs/decorators.py
def background_task(queue: str = "default"):
    def wrapper(func):
        func.__cqrs_background_task__ = True
        func.__cqrs_queue__ = queue
        func.__cqrs_task_name__ = f"{func.__module__}.{func.__qualname__}"
        return func
    return wrapper

from hexcore.domain.cqrs.decorators import background_task

@background_task(queue="reports")
async def generate_monthly_report(month: int, year: int) -> None:
    await build_report_pdf(month, year)

How the bus detects routing

Inside InMemoryCommandBus.dispatch(), the routing check is:

# hexcore/application/cqrs/in_memory_buses.py (simplified)
async def dispatch(self, command: Command) -> t.Any:
    cmd_type = type(command)
    is_background = getattr(cmd_type, "__cqrs_background__", False)

    if is_background:
        if not self._enqueuer or not self._serializer:
            raise RuntimeError(
                f"Command '{cmd_type.__name__}' requires background execution "
                "but no enqueuer or serializer is configured."
            )
        queue_name = getattr(cmd_type, "__cqrs_queue__", "default")
        payload = self._serializer.serialize(command)
        await self._enqueuer.enqueue_command(cmd_type.__name__, payload, queue=queue_name)
        return None

    # Standard local execution
    handler = self._registry.resolve_command_handler(cmd_type)

    async def final_handler(cmd: t.Any) -> t.Any:
        return await handler.handle(cmd)

    return await self._pipeline.execute(command, final_handler)

`ITaskEnqueuer` interface

The ITaskEnqueuer port (in hexcore.domain.cqrs.task_queues) defines four enqueue methods that any adapter must implement:

class ITaskEnqueuer(abc.ABC):
    @abc.abstractmethod
    async def enqueue_command(
        self, command_name: str, payload: dict, queue: str
    ) -> None: ...

    @abc.abstractmethod
    async def enqueue_event(
        self, event_name: str, payload: dict, queue: str
    ) -> None: ...

    @abc.abstractmethod
    async def enqueue_handler(
        self, handler_name: str, payload: dict, queue: str
    ) -> None: ...

    @abc.abstractmethod
    async def enqueue_task(
        self, task_name: str, payload: dict, queue: str
    ) -> None: ...

Worker-side: `CQRSConsumer`

On the worker side, CQRSConsumer receives the serialized payload and dispatches it to the local in-memory buses, which have the actual handlers registered.

# hexcore/infrastructure/workers/consumer.py (simplified)
class CQRSConsumer:
    def __init__(
        self,
        command_bus: AbstractCommandBus,
        event_bus: AbstractEventBus,
        serializer: ISerializer,
    ) -> None: ...

    async def process_command(self, payload: dict) -> None: ...
    async def process_event(self, payload: dict) -> None: ...
    async def process_handler(self, handler_name: str, payload: dict) -> None: ...
    async def process_task(self, task_name: str, payload: dict) -> None: ...

Complete wiring example with Celery

Define a background command

# app/commands.py
from hexcore.domain.cqrs.commands import Command
from hexcore.domain.cqrs.decorators import background_command

@background_command(queue="emails")
class SendWelcomeEmailCommand(Command):
    user_id: str
    to_address: str

Implement the handler

# app/handlers.py
from hexcore.domain.cqrs.handlers import AbstractCommandHandler
from app.commands import SendWelcomeEmailCommand

class SendWelcomeEmailHandler(AbstractCommandHandler[SendWelcomeEmailCommand, None]):
    async def handle(self, command: SendWelcomeEmailCommand) -> None:
        # This runs inside the Celery worker, not in the web process
        await send_email(command.to_address, "Welcome!")

Create the application-side bus (web process)

The application bus needs an enqueuer and serializer to send tasks to Celery. The handler does not need to be registered here — it is registered in the worker.

# app/buses.py
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus
from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer

# Application-side registry has NO handler for background commands
app_registry = HandlerRegistry()

# Build the enqueuer (see celery_app.py below)
from app.celery_app import celery_enqueuer

serializer = PydanticSerializer()
command_bus = InMemoryCommandBus(
    registry=app_registry,
    enqueuer=celery_enqueuer,
    serializer=serializer,
)

Set up a Celery enqueuer

# app/celery_app.py
from celery import Celery
from hexcore.domain.cqrs.task_queues import ITaskEnqueuer
import asyncio

celery_app = Celery("myapp", broker="redis://localhost:6379/0")

class CeleryEnqueuer(ITaskEnqueuer):
    async def enqueue_command(self, command_name: str, payload: dict, queue: str) -> None:
        celery_app.send_task(
            "hexcore.process_command",
            kwargs={"payload": payload},
            queue=queue,
        )

    async def enqueue_event(self, event_name: str, payload: dict, queue: str) -> None:
        celery_app.send_task(
            "hexcore.process_event",
            kwargs={"payload": payload},
            queue=queue,
        )

    async def enqueue_handler(self, handler_name: str, payload: dict, queue: str) -> None:
        celery_app.send_task(
            "hexcore.process_handler",
            kwargs={"handler_name": handler_name, "payload": payload},
            queue=queue,
        )

    async def enqueue_task(self, task_name: str, payload: dict, queue: str) -> None:
        celery_app.send_task(
            "hexcore.process_task",
            kwargs={"task_name": task_name, "payload": payload},
            queue=queue,
        )

celery_enqueuer = CeleryEnqueuer()

Set up the worker-side registry and CQRSConsumer

# app/worker.py
from celery import Celery
from hexcore.application.cqrs.registry import HandlerRegistry
from hexcore.application.cqrs.in_memory_buses import InMemoryCommandBus, InMemoryEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
from hexcore.infrastructure.workers.consumer import CQRSConsumer
from app.commands import SendWelcomeEmailCommand
from app.handlers import SendWelcomeEmailHandler

# Worker-side registry has the actual handlers
worker_registry = HandlerRegistry()
worker_registry.register_command_handler(
    SendWelcomeEmailCommand, SendWelcomeEmailHandler()
)

serializer = PydanticSerializer()
worker_command_bus = InMemoryCommandBus(registry=worker_registry)
worker_event_bus   = InMemoryEventBus()

consumer = CQRSConsumer(
    command_bus=worker_command_bus,
    event_bus=worker_event_bus,
    serializer=serializer,
)

celery_app = Celery("myapp", broker="redis://localhost:6379/0")

@celery_app.task(name="hexcore.process_command")
def process_command(payload: dict):
    import asyncio
    asyncio.run(consumer.process_command(payload))

@celery_app.task(name="hexcore.process_handler")
def process_handler(handler_name: str, payload: dict):
    import asyncio
    asyncio.run(consumer.process_handler(handler_name, payload))

@celery_app.task(name="hexcore.process_task")
def process_task(task_name: str, payload: dict):
    import asyncio
    asyncio.run(consumer.process_task(task_name, payload))

Dispatch from your web handler

# app/routers/auth.py
from fastapi import APIRouter, Depends
from hexcore.domain.cqrs.buses import ICommandBus
from app.buses import command_bus
from app.commands import SendWelcomeEmailCommand

router = APIRouter()

@router.post("/register")
async def register(user_id: str, email: str):
    # Returns None immediately — email is sent in background worker
    await command_bus.dispatch(
        SendWelcomeEmailCommand(user_id=user_id, to_address=email)
    )
    return {"status": "registered"}

Using `@background_handler` on event subscribers

from hexcore.domain.cqrs.decorators import background_handler
from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
from app.events import OrderShippedEvent
from app.celery_app import celery_enqueuer
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer

@background_handler(queue="notifications")
async def notify_customer_sms(event: OrderShippedEvent) -> None:
    await send_sms(event.customer_phone, f"Your order {event.order_id} shipped!")

# This handler runs in-process (synchronous)
async def update_order_status(event: OrderShippedEvent) -> None:
    await order_repo.mark_shipped(event.order_id)

event_bus = InMemoryEventBus(
    enqueuer=celery_enqueuer,
    serializer=PydanticSerializer(),
)
event_bus.subscribe(OrderShippedEvent, notify_customer_sms)   # → background
event_bus.subscribe(OrderShippedEvent, update_order_status)   # → in-process

@background_handler embeds func.__cqrs_handler_name__ as the fully-qualified module path (e.g. "app.events.notify_customer_sms"). The worker resolves this string via importlib at execution time, so the function must be importable from the worker process.

Decorator attributes reference

Decorator	Sets on target	Value
`@background_command(queue)`	`cls.__cqrs_background__`	`True`
`@background_command(queue)`	`cls.__cqrs_queue__`	queue name string
`@background_handler(queue)`	`func.__cqrs_background_handler__`	`True`
`@background_handler(queue)`	`func.__cqrs_queue__`	queue name string
`@background_handler(queue)`	`func.__cqrs_handler_name__`	`"module.qualname"`
`@background_task(queue)`	`func.__cqrs_background_task__`	`True`
`@background_task(queue)`	`func.__cqrs_queue__`	queue name string
`@background_task(queue)`	`func.__cqrs_task_name__`	`"module.qualname"`

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Route CQRS Messages to Background Workers Automatically

The three routing decorators

`@background_command(queue="...")`

`@background_handler(queue="...")`

`@background_task(queue="...")`

How the bus detects routing

`ITaskEnqueuer` interface

Worker-side: `CQRSConsumer`

Complete wiring example with Celery

Using `@background_handler` on event subscribers

Decorator attributes reference

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Documentation Index

​The three routing decorators

​@background_command(queue="...")

​@background_handler(queue="...")

​@background_task(queue="...")

​How the bus detects routing

​ITaskEnqueuer interface

​Worker-side: CQRSConsumer

​Complete wiring example with Celery

​Using @background_handler on event subscribers

​Decorator attributes reference

Build docs developers (and LLMs) love

The three routing decorators

`@background_command(queue="...")`

`@background_handler(queue="...")`

`@background_task(queue="...")`

How the bus detects routing

`ITaskEnqueuer` interface

Worker-side: `CQRSConsumer`

Complete wiring example with Celery

Using `@background_handler` on event subscribers

Decorator attributes reference