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 EventBus is the publish-subscribe backbone that decouples domain event producers from their consumers. A single publish() call fans the event out to every subscribed handler, regardless of whether those handlers run in-process or in remote workers. HexCore ships four implementations — in-memory for testing and simple deployments, and Redis Streams, PostgreSQL LISTEN/NOTIFY, and RabbitMQ for distributed production workloads.
The AbstractEventBus port
All implementations satisfy the AbstractEventBus interface defined in hexcore.domain.cqrs.buses:
# hexcore/domain/cqrs/buses.py (simplified)
import abc
import typing as t
from hexcore.domain.events import DomainEvent
class AbstractEventBus(abc.ABC):
@abc.abstractmethod
def subscribe(
self,
event_type: type[DomainEvent],
handler: t.Callable[[DomainEvent], t.Awaitable[None]],
) -> None:
"""Register a handler for an event type."""
raise NotImplementedError
@abc.abstractmethod
async def publish(self, event: DomainEvent) -> None:
"""Publish an event to all subscribed handlers."""
raise NotImplementedError
# Alias
IEventBus = AbstractEventBus
DomainEvent base class and fields
DomainEvent is defined in hexcore.domain.events. All domain events inherit from it and gain these auto-populated fields:
Auto-generated unique identifier for this event instance. Useful for deduplication and tracing.
UTC timestamp set at the moment the event is constructed.
Derived from the class name by stripping the "Event" suffix and uppercasing. For example, UserCreatedEvent → "USERCREATED". Used as the routing key in RabbitMQ and Redis.
Defining a domain event
from hexcore.domain.events import DomainEvent
class OrderPlacedEvent(DomainEvent):
order_id: str
customer_id: str
total_amount: float
Four implementations
InMemory
Redis Streams
PostgreSQL LISTEN/NOTIFY
RabbitMQ
The simplest implementation — all handlers run in the same process. Use this for local development, unit tests, and applications that don’t need distributed event delivery.from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
from app.events import OrderPlacedEvent
bus = InMemoryEventBus()
# Subscribe a handler
async def on_order_placed(event: OrderPlacedEvent) -> None:
print(f"New order: {event.order_id}")
bus.subscribe(OrderPlacedEvent, on_order_placed)
# Publish
await bus.publish(OrderPlacedEvent(
order_id="ord-1",
customer_id="cust-42",
total_amount=49.99,
))
InMemoryEventBus also supports Smart Routing — if a subscribed handler is decorated with @background_handler, the bus will forward the event payload to the configured ITaskEnqueuer instead of calling the handler directly.from hexcore.application.cqrs.in_memory_buses import InMemoryEventBus
bus = InMemoryEventBus(
enqueuer=my_celery_enqueuer,
serializer=pydantic_serializer,
)
RedisEventBus publishes events via XADD and consumes them through Redis Consumer Groups (XREADGROUP / XACK). Events are persisted in the stream until acknowledged, so slow consumers don’t lose messages.import redis.asyncio as aioredis
from hexcore.infrastructure.cqrs.redis_bus import RedisEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
redis_client = aioredis.from_url("redis://localhost:6379")
serializer = PydanticSerializer()
bus = RedisEventBus(
redis_client=redis_client,
serializer=serializer,
stream_name="hexcore:events",
group_name="order-service",
consumer_name="worker-1", # optional, auto-generated if omitted
)
Constructor parameters
redis_client
redis.asyncio.Redis
required
An active async Redis connection.
serializer
PydanticSerializer
required
Serializer used to encode/decode event payloads to/from JSON.
Name of the Redis Stream key (e.g. "hexcore:events").
Consumer group name. Multiple replicas sharing the same group receive each message only once.
consumer_name
str
default:"consumer-<uuid>"
Unique name for this consumer within the group. Auto-generated when omitted.
enqueuer
ITaskEnqueuer | None
default:"None"
Optional enqueuer for Smart Routing of @background_handler-decorated subscribers.
Consuming events
from app.events import OrderPlacedEvent
async def on_order_placed(event: OrderPlacedEvent) -> None:
await send_confirmation_email(event.customer_id)
bus.subscribe(OrderPlacedEvent, on_order_placed)
# Start the blocking consumer loop (run in an asyncio.Task or background service)
await bus.start_consuming()
# Graceful shutdown
await bus.stop()
PostgresEventBus uses PostgreSQL’s built-in LISTEN/NOTIFY mechanism via asyncpg. It requires no additional infrastructure if you already use PostgreSQL, and is ideal for low-to-medium event volumes within a single database.import asyncpg
from hexcore.infrastructure.cqrs.postgres_bus import PostgresEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")
serializer = PydanticSerializer()
bus = PostgresEventBus(
pool=pool,
serializer=serializer,
channel_name="hexcore_events",
)
Constructor parameters
An active asyncpg connection pool.
serializer
PydanticSerializer
required
Serializer used to JSON-encode/decode event payloads.
PostgreSQL notification channel name (e.g. "hexcore_events").
enqueuer
ITaskEnqueuer | None
default:"None"
Optional enqueuer for Smart Routing of background handlers.
Publishing and consuming
from app.events import OrderPlacedEvent
async def on_order_placed(event: OrderPlacedEvent) -> None:
await update_analytics(event.order_id)
bus.subscribe(OrderPlacedEvent, on_order_placed)
# Acquires a dedicated persistent connection and blocks on LISTEN
await bus.start_consuming()
# Graceful shutdown
await bus.stop()
PostgreSQL LISTEN/NOTIFY does not persist notifications. If no listener is connected when publish() is called, or if the connection drops, the notification is permanently lost. Do not use PostgresEventBus for events that must be reliably delivered — use Redis Streams or RabbitMQ instead.
RabbitMQEventBus uses aio-pika and a durable topic exchange to route events across services. Events are published as persistent AMQP messages with the event_name as the routing key.import aio_pika
from hexcore.infrastructure.cqrs.rabbitmq import RabbitMQEventBus
from hexcore.infrastructure.cqrs.pydantic_serializer import PydanticSerializer
connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
serializer = PydanticSerializer()
bus = RabbitMQEventBus(
connection=connection,
serializer=serializer,
exchange_name="hexcore.events",
queue_name="order-service.queue",
)
Constructor parameters
connection
AbstractRobustConnection
required
A robust aio-pika connection. Robust connections automatically reconnect on failure.
serializer
AbstractSerializer
required
Serializer for encoding/decoding event payloads.
pipeline
MiddlewarePipeline | None
default:"None"
Optional middleware pipeline applied to each consumed message.
exchange_name
str
default:"hexcore.events"
Name of the durable topic exchange. All producers and consumers must use the same exchange.
queue_name
str
default:"hexcore.worker.queue"
Durable queue name for this service/worker. Each microservice should use a distinct queue.
Publishing and consuming
from app.events import OrderPlacedEvent
async def on_order_placed(event: OrderPlacedEvent) -> None:
await notify_warehouse(event.order_id)
# Subscribe before starting consumption
bus.subscribe(OrderPlacedEvent, on_order_placed)
# Publish from any coroutine
await bus.publish(OrderPlacedEvent(
order_id="ord-5",
customer_id="cust-99",
total_amount=199.00,
))
# Start consuming (binds queue to exchange for every subscribed event type)
await bus.start_consuming()
Subscribing multiple handlers to one event
You can subscribe as many handlers as needed to a single event type. All handlers are called for each published event.
bus.subscribe(OrderPlacedEvent, send_confirmation_email)
bus.subscribe(OrderPlacedEvent, update_inventory)
bus.subscribe(OrderPlacedEvent, log_audit_trail)
Lifecycle in FastAPI
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio
from app.events import OrderPlacedEvent
from app.handlers import on_order_placed
from app.bus import event_bus # your configured bus instance
@asynccontextmanager
async def lifespan(app: FastAPI):
event_bus.subscribe(OrderPlacedEvent, on_order_placed)
consuming_task = asyncio.create_task(event_bus.start_consuming())
yield
await event_bus.stop()
consuming_task.cancel()
app = FastAPI(lifespan=lifespan)
InMemoryEventBus has no start_consuming() method — it executes handlers synchronously inline. Only Redis, Postgres, and RabbitMQ buses require an explicit consumer loop.