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.

ITaskEnqueuer is HexCore’s output port for delegating work to an external task queue. When a command or event handler is decorated with @background_command or @background_handler, the appropriate bus dispatches the serialised payload to the enqueuer instead of executing the handler inline. Two concrete adapters are included: CeleryEnqueuer (sync Celery, run via asyncio.to_thread) and ProcrastinateEnqueuer (async PostgreSQL-backed).

​
ITaskEnqueuer

hexcore.domain.cqrs.task_queues.ITaskEnqueuer Abstract base class (output port) that all task queue adapters 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: ...

​
Abstract methods

​
enqueue_command(command_name, payload, queue)
Awaitable[None]
Enqueues a serialised domain command for background execution. command_name is the qualified command class name (e.g. "CreateUserCommand"). payload is the pre-serialised dict produced by ISerializer.serialize().
​
enqueue_event(event_name, payload, queue)
Awaitable[None]
Enqueues a serialised domain event for generic distribution. Both CeleryEnqueuer and ProcrastinateEnqueuer provide a no-op default; use enqueue_handler for targeted delivery.
​
enqueue_handler(handler_name, payload, queue)
Awaitable[None]
Enqueues a specific event handler for background execution. handler_name is the handler’s __cqrs_handler_name__ attribute (its qualified import path). This is the primary method used by Smart Routing.
​
enqueue_task(task_name, payload, queue)
Awaitable[None]
Enqueues a generic background task decorated with @background_task. task_name is the task’s __cqrs_task_name__ attribute. payload is passed as **kwargs to the task function.

​
CeleryEnqueuer

hexcore.infrastructure.task_queues.celery_adapter.CeleryEnqueuer Adapts ITaskEnqueuer to Celery’s synchronous send_task API. Because ITaskEnqueuer is fully async and Celery’s send_task blocks, each call uses asyncio.to_thread to avoid freezing the event loop in FastAPI or other async frameworks.

​
Constructor

CeleryEnqueuer(app: "Celery")
​
app
celery.Celery
required
A configured Celery application instance. The enqueuer calls app.send_task(name, kwargs=..., queue=...).

​
Method implementations

MethodCelery task nameKwargs sent
enqueue_commandhexcore.process_command{"payload": payload}
enqueue_event(no-op)—
enqueue_handlerhexcore.process_handler{"handler_name": handler_name, "payload": payload}
enqueue_taskhexcore.process_task{"task_name": task_name, "payload": payload}

​
Example

from celery import Celery
from hexcore.infrastructure.task_queues.celery_adapter import CeleryEnqueuer

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

# Used automatically by InMemoryCommandBus / InMemoryEventBus Smart Routing,
# or called directly:
await enqueuer.enqueue_command("CreateUserCommand", payload, queue="default")

​
register_hexcore_celery_tasks()

hexcore.infrastructure.task_queues.celery_adapter.register_hexcore_celery_tasks 
def register_hexcore_celery_tasks(app: "Celery", consumer: "CQRSConsumer") -> None
Auto-registers the three standard HexCore Celery tasks on a Celery application instance. Call this in your celery.py worker module so that the worker knows how to handle messages produced by CeleryEnqueuer.
​
app
celery.Celery
required
The Celery application to register tasks on.
​
consumer
CQRSConsumer
required
The CQRSConsumer instance wired with the local in-memory buses and serialiser. Its process_* methods are called synchronously via asyncio.run() inside each task.

​
Registered tasks

Task nameCelery task signatureDelegates to
hexcore.process_commandprocess_command(payload: dict)consumer.process_command(payload)
hexcore.process_handlerprocess_handler(handler_name: str, payload: dict)consumer.process_handler(handler_name, payload)
hexcore.process_taskprocess_task(task_name: str, payload: dict)consumer.process_task(task_name, payload)
# celery_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
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
from myapp.cqrs import command_bus, event_bus

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

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

register_hexcore_celery_tasks(celery_app, consumer)

​
ProcrastinateEnqueuer

hexcore.infrastructure.task_queues.procrastinate_adapter.ProcrastinateEnqueuer Adapts ITaskEnqueuer to Procrastinate’s fully-async defer_async API. No thread offloading is required — each method is a native coroutine.

​
Constructor

ProcrastinateEnqueuer(app: "procrastinate.App")
​
app
procrastinate.App
required
A configured procrastinate.App instance (connected to PostgreSQL via asyncpg). The enqueuer uses app.configure_task(name=..., queue=...).defer_async(...).

​
Method implementations

MethodTask nameArgs deferred
enqueue_commandhexcore.process_commandpayload=payload
enqueue_event(no-op)—
enqueue_handlerhexcore.process_handlerhandler_name=handler_name, payload=payload
enqueue_taskhexcore.process_tasktask_name=task_name, payload=payload

​
register_hexcore_procrastinate_tasks()

def register_hexcore_procrastinate_tasks(app: "procrastinate.App", consumer: "CQRSConsumer") -> None
Registers the three standard HexCore tasks as native async Procrastinate tasks. Call in your worker entrypoint.
Task nameAsync task signature
hexcore.process_commandprocess_command(payload: dict)
hexcore.process_handlerprocess_handler(handler_name: str, payload: dict)
hexcore.process_taskprocess_task(task_name: str, payload: dict)
# procrastinate_worker.py
import procrastinate
from hexcore.infrastructure.task_queues.procrastinate_adapter import (
    ProcrastinateEnqueuer,
    register_hexcore_procrastinate_tasks,
)
from hexcore.infrastructure.workers.consumer import CQRSConsumer
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
from myapp.cqrs import command_bus, event_bus

pg_connector = procrastinate.AiopgConnector(dsn="postgresql://user:pass@localhost/mydb")
proc_app = procrastinate.App(connector=pg_connector)

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

register_hexcore_procrastinate_tasks(proc_app, consumer)

​
CQRSConsumer

hexcore.infrastructure.workers.consumer.CQRSConsumer Universal CQRS message consumer for worker processes. Receives serialised payloads from any task queue backend, deserialises them using the configured ISerializer, and dispatches the resulting commands or events to local in-memory buses. Runs inside the worker process, which maintains its own handler registry.

​
Constructor

CQRSConsumer(
    command_bus: AbstractCommandBus,
    event_bus: AbstractEventBus,
    serializer: ISerializer,
)
​
command_bus
AbstractCommandBus
required
The local command bus (typically InMemoryCommandBus) with all command handlers registered. Receives deserialised commands via dispatch().
​
event_bus
AbstractEventBus
required
The local event bus (typically InMemoryEventBus). Receives deserialised events via publish().
​
serializer
ISerializer
required
The serialiser instance (e.g. PydanticSerializer) used to reconstruct domain objects from raw dict payloads.

​
Methods

​
process_command(payload: dict[str, Any])
Awaitable[None]
Deserialises payload into a Command instance and dispatches it to the local command_bus. Logs and re-raises on failure.
​
process_event(payload: dict[str, Any])
Awaitable[None]
Deserialises payload into a DomainEvent instance and publishes it to the local event_bus. All locally subscribed handlers for that event type are invoked.
​
process_handler(handler_name: str, payload: dict[str, Any])
Awaitable[None]
Deserialises payload into a DomainEvent, then resolves handler_name as a Python qualified import path (e.g. "myapp.events.on_user_created") and calls it directly. This is the target of enqueue_handler Smart Routing — only the single decorated handler runs, not all subscribers.
​
process_task(task_name: str, payload: dict[str, Any])
Awaitable[None]
Resolves task_name as a qualified import path and calls it with **payload as keyword arguments. Used by enqueue_task for generic background functions.

​
PydanticSerializer

hexcore.infrastructure.cqrs.pydantic_serializer.PydanticSerializer Default ISerializer implementation for HexCore CQRS. Serialises any Pydantic BaseModel subclass to a dict envelope carrying the fully-qualified type name and reconstructs it on the other end using model_validate.

​
Wire format

{
  "__type__": "myapp.commands.CreateUserCommand",
  "__data__": { "user_id": "...", "email": "..." }
}

​
Methods

​
serialize(message: Any)
dict[str, Any]
Produces the wire envelope. __type__ is f"{module}.{qualname}". __data__ is message.model_dump(mode="json").
serializer = PydanticSerializer()
payload = serializer.serialize(CreateUserCommand(email="alice@example.com"))
# {"__type__": "myapp.commands.CreateUserCommand", "__data__": {"email": "alice@example.com"}}
​
deserialize(data: dict[str, Any])
Any
Resolves __type__ via importlib.import_module + getattr, then reconstructs the object with cls.model_validate(__data__). Raises DeserializationError if __type__ or __data__ are missing, the type cannot be imported, or validation fails.
command = serializer.deserialize(payload)
assert isinstance(command, CreateUserCommand)
Both ends of the serialisation boundary must have the command/event class importable at the same qualified path. Renaming or moving classes without re-deploying workers will cause DeserializationError.

Build docs developers (and LLMs) love

Get started for freeTalk to us