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.

HexCore’s task-queue integration is built around a single interface — ITaskEnqueuer — that your command bus, event bus, and scheduler call to offload work to a background worker. Two adapter implementations are provided out of the box: CeleryEnqueuer for Redis/RabbitMQ-backed queues, and ProcrastinateEnqueuer for PostgreSQL-native queuing. This guide covers both adapters, the universal CQRSConsumer that runs on the worker side, and a complete wiring example.

​
The ITaskEnqueuer Interface

import abc
import typing as t


class ITaskEnqueuer(abc.ABC):
    @abc.abstractmethod
    async def enqueue_command(
        self, command_name: str, payload: dict[str, t.Any], queue: str
    ) -> None:
        """Enqueue a command for background execution."""

    @abc.abstractmethod
    async def enqueue_handler(
        self, handler_name: str, payload: dict[str, t.Any], queue: str
    ) -> None:
        """Enqueue a specific event handler decorated with @background_handler."""

    @abc.abstractmethod
    async def enqueue_event(
        self, event_name: str, payload: dict[str, t.Any], queue: str
    ) -> None:
        """Broadcast an event to all subscribers (rarely used directly)."""

    @abc.abstractmethod
    async def enqueue_task(
        self, task_name: str, payload: dict[str, t.Any], queue: str
    ) -> None:
        """Enqueue a generic background task decorated with @background_task."""
​
enqueue_command
async (command_name, payload, queue) -> None
Called automatically by InMemoryCommandBus when the command class is decorated with @background_command. Sends the serialised command payload to the worker.
​
enqueue_handler
async (handler_name, payload, queue) -> None
Called by the event bus when an event handler is decorated with @background_handler. Routes a specific handler function by its qualified name.
​
enqueue_event
async (event_name, payload, queue) -> None
Broadcasts an event payload. Usually left as a pass stub — prefer @background_handler for targeted routing.
​
enqueue_task
async (task_name, payload, queue) -> None
Called when a generic async function decorated with @background_task is enqueued. The payload dict becomes the **kwargs of the task function on the worker.

​
Adapters

​
CeleryEnqueuer

CeleryEnqueuer wraps celery.Celery.send_task (a synchronous call) inside asyncio.to_thread so it never blocks the FastAPI event loop. Pass your Celery app instance to the constructor:
src/infrastructure/task_queues/celery_setup.py
from celery import Celery
from hexcore.infrastructure.task_queues.celery_adapter import CeleryEnqueuer

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

enqueuer = CeleryEnqueuer(app=celery_app)
Under the hood each method calls app.send_task with a fixed HexCore task name:
MethodCelery task name
enqueue_commandhexcore.process_command
enqueue_handlerhexcore.process_handler
enqueue_taskhexcore.process_task

​
register_hexcore_celery_tasks

Call this on the worker side to register the three task routes in the Celery app. It wires CQRSConsumer methods as Celery task handlers using asyncio.run():
worker.py
from celery import Celery
from hexcore.infrastructure.task_queues.celery_adapter import (
    CeleryEnqueuer,
    register_hexcore_celery_tasks,
)
from hexcore.infrastructure.workers.consumer import CQRSConsumer

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

# Build your local buses (with handlers registered)
from src.infrastructure.buses import command_bus, event_bus, serializer

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

# Register hexcore.process_command, hexcore.process_handler, hexcore.process_task
register_hexcore_celery_tasks(celery_app, consumer)
Start the worker:
celery -A worker worker --loglevel=info -Q default,high_priority

​
CQRSConsumer

CQRSConsumer is the universal worker-side dispatcher. It receives serialised payloads from the queue backend, deserialises them, and routes them to the correct local bus or handler. 
from hexcore.infrastructure.workers.consumer import CQRSConsumer

consumer = CQRSConsumer(
    command_bus=command_bus,   # AbstractCommandBus with handlers registered
    event_bus=event_bus,       # AbstractEventBus with subscribers registered
    serializer=serializer,     # ISerializer (e.g. PydanticSerializer)
)
​
command_bus
AbstractCommandBus
The in-memory command bus with all handlers registered. Receives deserialised Command objects.
​
event_bus
AbstractEventBus
The in-memory event bus with all subscribers registered. Used by process_event.
​
serializer
ISerializer
Deserialises raw dict payloads back into typed Command or DomainEvent objects.

​
Consumer methods

MethodCalled byDescription
process_command(payload)hexcore.process_commandDeserialises and dispatches a command to command_bus
process_event(payload)(manual)Deserialises and publishes an event to event_bus
process_handler(handler_name, payload)hexcore.process_handlerResolves handler_name by qualified import path and calls it with the event
process_task(task_name, payload)hexcore.process_taskResolves task_name and calls it with **payload as kwargs

​
Complete Worker Entrypoint

The following example shows a full Celery worker file that registers HexCore tasks and initialises all database connections on startup: 
import asyncio
from celery import Celery
from celery.signals import worker_process_init

from hexcore.infrastructure.task_queues.celery_adapter import (
    CeleryEnqueuer,
    register_hexcore_celery_tasks,
)
from hexcore.infrastructure.workers.consumer import CQRSConsumer
from hexcore.infrastructure.repositories.orms.sqlalchemy.session import (
    reset_sqlalchemy_engine,
)

# --- App setup ---
celery_app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

# --- Reset SQLAlchemy pool when the worker process starts ---
@worker_process_init.connect
def init_worker(**kwargs):
    asyncio.run(reset_sqlalchemy_engine())

# --- Local buses (handlers registered) ---
from src.infrastructure.buses import build_command_bus, build_event_bus
from src.infrastructure.serializer import serializer

command_bus = build_command_bus()
event_bus = build_event_bus()

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

# --- Wire HexCore task routes ---
register_hexcore_celery_tasks(celery_app, consumer)

​
Enqueuing Tasks with Smart Routing Decorators

Use the CQRS decorators to mark which commands, handlers, and tasks should run in the background. HexCore reads these annotations at runtime to call the correct ITaskEnqueuer method automatically:
src/domain/notifications/commands.py
from hexcore.domain.cqrs.decorators import background_command, background_handler, background_task
from hexcore.domain.cqrs.commands import Command
from src.domain.users.events import UserCreatedEvent


# Entire command goes to a background worker
@background_command(queue="notifications")
class SendWelcomeEmailCommand(Command):
    user_id: str
    email: str


# Only this handler runs in background; other handlers for the same event run inline
@background_handler(queue="analytics")
async def track_signup_analytics(event: UserCreatedEvent) -> None:
    # expensive analytics call
    ...


# Generic background task not tied to a command or event
@background_task(queue="maintenance")
async def purge_expired_sessions(days: int) -> None:
    ...
To enqueue the generic task manually: 
await enqueuer.enqueue_task(
    task_name=purge_expired_sessions.__cqrs_task_name__,
    payload={"days": 30},
    queue=purge_expired_sessions.__cqrs_queue__,
)
The @background_command and @background_handler decorators attach __cqrs_task_name__ and __cqrs_queue__ attributes to the decorated class or function. HexCore’s buses read these automatically — you never call enqueue_command or enqueue_handler directly.

Build docs developers (and LLMs) love

Get started for freeTalk to us