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.

Domain events are the primary mechanism by which one part of your domain communicates a meaningful state change to the rest of the system — without creating direct coupling. All events in HexCore are immutable Pydantic models (frozen=True) stamped with a UUID and a UTC timestamp on construction. Three concrete subclasses cover the standard entity lifecycle, and the EventBus abstract class defines the port your infrastructure must implement for publishing and subscribing.

DomainEvent

from hexcore.domain.events import DomainEvent
class DomainEvent(BaseModel):
    model_config = ConfigDict(from_attributes=True, frozen=True)
The base class for all domain events. Subclass it to describe every significant occurrence in your domain.

Fields

event_id
UUID
default:"uuid4()"
Unique identifier for this specific event instance. Auto-generated.
occurred_on
datetime
UTC timestamp of when the event occurred, set automatically via datetime.now(UTC).

Computed property

event_name

@computed_field
@property
def event_name(self) -> str
Returns a canonical string identifier for the event type. The class name is stripped of the trailing "Event" suffix and converted to uppercase.
event_name
str
The class name with "Event" removed, uppercased. For example, UserCreatedEvent"USERCREATED".
event_name is a Pydantic computed_field, so it is included in .model_dump() and serialised in JSON output automatically.

Generic Entity Lifecycle Events

HexCore provides three generic event types that cover the standard CRUD lifecycle of any entity.

EntityCreatedEvent[T]

from hexcore.domain.events import EntityCreatedEvent

class EntityCreatedEvent(DomainEvent, Generic[T]):
    entity_id: UUID
    entity_data: T
Raised when a new entity is successfully created. The full entity snapshot is carried in entity_data so downstream handlers do not need to re-query the database.
entity_id
UUID
required
The id of the entity that was created.
entity_data
T
required
The full entity instance. T must be bound to BaseEntity.

EntityUpdatedEvent[T]

from hexcore.domain.events import EntityUpdatedEvent

class EntityUpdatedEvent(DomainEvent, Generic[T]):
    entity_id: UUID
    entity_data: T
Raised when an existing entity has been modified. Carries the post-update snapshot.
entity_id
UUID
required
The id of the entity that was updated.
entity_data
T
required
The entity instance reflecting the new state after the update.

EntityDeletedEvent

from hexcore.domain.events import EntityDeletedEvent

class EntityDeletedEvent(DomainEvent):
    entity_id: UUID
Raised when an entity is removed (either hard-deleted or logically deactivated). Only the identifier is carried, since the entity data may no longer be accessible.
entity_id
UUID
required
The id of the entity that was deleted.

EventHandler type alias

EventHandler = Callable[[DomainEvent], Awaitable[None]]
A type alias for any async callable that accepts a single DomainEvent argument and returns nothing. Use this type when annotating subscriber functions passed to EventBus.subscribe().

EventBus

from hexcore.domain.events import EventBus
An abstract class that defines the port for publishing and subscribing to domain events. Infrastructure adapters (e.g., InMemoryEventBus, a Redis-backed bus) must implement this interface.

Methods

subscribe

@abstractmethod
def subscribe(self, event_type: type, handler: EventHandler) -> None
Registers a handler callable for a specific event type. Multiple handlers may be subscribed to the same event type; each will be invoked when the event is published.
event_type
type
required
The event class to subscribe to (e.g., UserCreatedEvent).
handler
EventHandler
required
An async callable matching Callable[[DomainEvent], Awaitable[None]].

publish

@abstractmethod
async def publish(self, event: Any) -> None
Publishes an event to all registered handlers for its type. Implementations must iterate over all subscribers and await each handler.
event
Any
required
The domain event instance to publish.

Deprecated aliases

The following methods are deprecated and will be removed in a future version. Migrate to subscribe() and publish() respectively.

register (deprecated)

def register(self, event_type: type, handler: EventHandler) -> None
Alias for subscribe(). Emits a DeprecationWarning at runtime.

dispatch (deprecated)

async def dispatch(self, event: Any) -> None
Alias for publish(). Emits a DeprecationWarning at runtime.

IEventDispatcher alias

IEventDispatcher = EventBus
A backward-compatibility alias for EventBus. Code that imports IEventDispatcher continues to work without changes, but new code should use EventBus directly.

Usage example

from uuid import UUID
from hexcore.domain.events import DomainEvent

class UserRegisteredEvent(DomainEvent):
    user_id: UUID
    email: str

Build docs developers (and LLMs) love