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.

Entities and domain events are the two core building blocks of HexCore’s domain model. BaseEntity provides identity, lifecycle timestamps, and an event registry so that your aggregates can raise business events without coupling to any infrastructure. DomainEvent and its typed subclasses give those events a consistent, immutable shape. This page explains all fields and methods, then walks through real examples.

​
BaseEntity

BaseEntity lives in hexcore.domain.base and is a Pydantic BaseModel configured for DDD usage — it supports ORM attribute reading (from_attributes=True) and re-validates fields on assignment (validate_assignment=True). 
from hexcore.domain.base import BaseEntity

​
Fields

​
id
UUID
default:"uuid4()"
Universally unique identifier for the entity. Generated automatically when the entity is created. Never changes after construction.
​
created_at
datetime
default:"datetime.now(UTC)"
UTC timestamp of when the entity was first created. Set automatically on construction.
​
updated_at
datetime
default:"datetime.now(UTC)"
UTC timestamp of the most recent update. Update this field in your entity methods when state changes.
​
is_active
bool | None
default:"True"
Soft-delete flag. Active entities have is_active=True. Calling deactivate() sets this to False without removing the record from the database.

​
Methods

​
register_event(event)
None
Appends a DomainEvent instance to the entity’s internal event list. Call this inside entity mutation methods to record that something meaningful happened.
self.register_event(UserCreatedEvent(entity_id=self.id, entity_data=self))
​
pull_domain_events()
List[DomainEvent]
Returns all registered events and clears the internal list. The Unit of Work calls this method to collect events before dispatching them. After calling pull_domain_events() the entity holds no pending events.
​
clear_domain_events()
None
Discards all pending events without returning them. Useful in tests when you want to reset state without dispatching anything.
​
deactivate()
Awaitable[None]
Async method that sets is_active = False. Represents a logical (soft) delete. The entity is not removed from the database — queries should filter by is_active=True to exclude deactivated entities.
await user.deactivate()

​
Defining an entity

from uuid import UUID
from datetime import datetime, UTC
from hexcore.domain.base import BaseEntity
from hexcore.domain.events import EntityCreatedEvent, EntityUpdatedEvent


class UserCreatedEvent(EntityCreatedEvent["User"]):
    pass


class UserEmailChangedEvent(EntityUpdatedEvent["User"]):
    previous_email: str


class User(BaseEntity):
    name: str
    email: str
    role: str = "member"

    @classmethod
    def create(cls, name: str, email: str) -> "User":
        user = cls(name=name, email=email)
        user.register_event(
            UserCreatedEvent(entity_id=user.id, entity_data=user)
        )
        return user

    def change_email(self, new_email: str) -> None:
        previous = self.email
        self.email = new_email
        self.updated_at = datetime.now(UTC)
        self.register_event(
            UserEmailChangedEvent(
                entity_id=self.id,
                entity_data=self,
                previous_email=previous,
            )
        )
Place entity mutation logic in named factory class methods (like User.create(...)) or regular methods (like change_email()), never directly in application code. This keeps domain invariants enforced at the entity boundary.

​
DomainEvent

DomainEvent lives in hexcore.domain.events. It is a frozen Pydantic model — all events are immutable once created. 
from hexcore.domain.events import DomainEvent

​
Fields

​
event_id
UUID
default:"uuid4()"
Unique identifier for this specific event occurrence. Auto-generated.
​
occurred_on
datetime
default:"datetime.now(UTC)"
UTC timestamp of when the event was raised. Auto-generated.
​
event_name
str
Computed property. Returns the class name with "Event" stripped and uppercased (e.g. UserCreatedEvent → "USERCREATED"). Used for serialisation and event routing.

​
Custom domain events

Subclass DomainEvent directly when you need an event that does not correspond to a standard entity lifecycle: 
from hexcore.domain.events import DomainEvent


class PasswordResetRequestedEvent(DomainEvent):
    user_id: UUID
    reset_token: str
    expires_at: datetime

​
Typed lifecycle events

HexCore provides three generic lifecycle events that carry the triggering entity.

​
EntityCreatedEvent[T]

from hexcore.domain.events import EntityCreatedEvent
​
entity_id
UUID
The id of the entity that was created.
​
entity_data
T
The full entity instance at the time of creation.

​
EntityUpdatedEvent[T]

from hexcore.domain.events import EntityUpdatedEvent
​
entity_id
UUID
The id of the updated entity.
​
entity_data
T
The full entity instance after the update.

​
EntityDeletedEvent

from hexcore.domain.events import EntityDeletedEvent
​
entity_id
UUID
The id of the deleted entity. Does not carry entity_data because the entity is no longer available.

​
EventBus

EventBus is the abstract port for publishing and subscribing to domain events. ServerConfig.event_bus holds the active implementation (defaults to InMemoryEventBus). 
from hexcore.domain.events import EventBus, EventHandler

class MyEventBus(EventBus):
    def subscribe(self, event_type: type, handler: EventHandler) -> None:
        ...

    async def publish(self, event) -> None:
        ...
Subscribe a handler and publish manually: 
from hexcore.config import LazyConfig

event_bus = LazyConfig.get_config().event_bus

async def on_user_created(event: UserCreatedEvent) -> None:
    print(f"New user: {event.entity_data.email}")

event_bus.subscribe(UserCreatedEvent, on_user_created)

# Publish directly (the UoW dispatches automatically via dispatch_events())
await event_bus.publish(UserCreatedEvent(entity_id=user.id, entity_data=user))

​
Complete example: entity → event → dispatch

The following example shows the full lifecycle: create an entity, register events, collect them from the entity, and dispatch them via the bus. 
import asyncio
from uuid import UUID
from datetime import datetime, UTC
from hexcore.domain.base import BaseEntity
from hexcore.domain.events import EntityCreatedEvent, EntityDeletedEvent
from hexcore.config import LazyConfig


# --- Domain layer ---

class OrderCreatedEvent(EntityCreatedEvent["Order"]):
    pass


class OrderCancelledEvent(EntityDeletedEvent):
    reason: str


class Order(BaseEntity):
    product_id: UUID
    quantity: int
    status: str = "pending"

    @classmethod
    def place(cls, product_id: UUID, quantity: int) -> "Order":
        order = cls(product_id=product_id, quantity=quantity)
        order.register_event(
            OrderCreatedEvent(entity_id=order.id, entity_data=order)
        )
        return order

    async def cancel(self, reason: str) -> None:
        self.status = "cancelled"
        self.updated_at = datetime.now(UTC)
        await self.deactivate()
        self.register_event(
            OrderCancelledEvent(entity_id=self.id, reason=reason)
        )


# --- Application layer ---

async def main() -> None:
    event_bus = LazyConfig.get_config().event_bus

    async def log_created(event: OrderCreatedEvent) -> None:
        print(f"Order placed: {event.entity_id}")

    async def log_cancelled(event: OrderCancelledEvent) -> None:
        print(f"Order cancelled ({event.entity_id}): {event.reason}")

    event_bus.subscribe(OrderCreatedEvent, log_created)
    event_bus.subscribe(OrderCancelledEvent, log_cancelled)

    order = Order.place(product_id=UUID("..."), quantity=3)

    # Pull events and dispatch them
    events = order.pull_domain_events()
    for event in events:
        await event_bus.publish(event)

    await order.cancel(reason="Customer request")

    events = order.pull_domain_events()
    for event in events:
        await event_bus.publish(event)


asyncio.run(main())
In practice you never dispatch events manually like this. The Unit of Work collects entities via collect_entity() and calls dispatch_events() automatically during commit(). See the Unit of Work guide for details.

Build docs developers (and LLMs) love

Get started for freeTalk to us