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-Driven Design (DDD) is a software development approach that places the domain model at the centre of the system and uses a shared language between developers and domain experts to express business rules in code. HexCore implements the essential DDD building blocks — entities, value objects, domain events, repositories, domain services, and the Unit of Work — as ready-to-extend Python abstractions built on Pydantic v2. This page explains each building block, shows you the real class signatures, and provides code examples you can adapt directly to your domain.

​
Entities

An entity is an object defined by its identity rather than its attributes. Two users with the same name are still different entities because they have different id values. HexCore’s BaseEntity is a Pydantic v2 BaseModel subclass that adds four universal fields plus built-in domain event management: 
# hexcore/domain/base.py
class BaseEntity(BaseModel):
    id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    is_active: Optional[bool] = True

    # Domain event management
    def register_event(self, event: DomainEvent) -> None: ...
    def pull_domain_events(self) -> List[DomainEvent]: ...
    def clear_domain_events(self) -> None: ...
    async def deactivate(self) -> None: ...  # sets is_active = False (soft delete)
Defining your own entity: 
from uuid import UUID
from hexcore.domain.base import BaseEntity

class User(BaseEntity):
    username: str
    email: str
    role: str = "viewer"

# Pydantic validates on construction and on every assignment (validate_assignment=True)
user = User(username="alice", email="alice@example.com")
print(user.id)          # UUID('...')
print(user.is_active)   # True
Soft-deleting an entity: 
await user.deactivate()
print(user.is_active)  # False
BaseEntity is configured with from_attributes=True, which lets you construct entities directly from ORM model instances using User.model_validate(orm_object).

​
Value Objects

A value object is an immutable descriptor with no conceptual identity — two Money(100, "USD") instances are the same regardless of which object reference you hold. In HexCore, model value objects as frozen Pydantic BaseModel subclasses: 
from pydantic import BaseModel

class Money(BaseModel):
    amount: float
    currency: str

    model_config = {"frozen": True}  # immutable

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

    model_config = {"frozen": True}
Embed them directly in your entities: 
class Order(BaseEntity):
    buyer_id: UUID
    shipping_address: Address
    total: Money
Frozen Pydantic models are hashable, so you can use them in sets or as dictionary keys — handy when comparing addresses or currencies.

​
Domain Events

A domain event represents something meaningful that has already happened in the domain. Events are facts: they are immutable and named in the past tense. HexCore provides four event classes:
ClassWhen to use
DomainEventBase for any custom domain event
EntityCreatedEvent[T]An entity was first created and persisted
EntityUpdatedEvent[T]An entity’s state changed
EntityDeletedEventAn entity was removed
# hexcore/domain/events.py
class DomainEvent(BaseModel):
    event_id: UUID = Field(default_factory=uuid4)
    occurred_on: datetime = Field(default_factory=lambda: datetime.now(UTC))

    @computed_field
    @property
    def event_name(self) -> str:
        return self.__class__.__name__.replace("Event", "").upper()

class EntityCreatedEvent(DomainEvent, Generic[T]):
    entity_id: UUID
    entity_data: T

class EntityUpdatedEvent(DomainEvent, Generic[T]):
    entity_id: UUID
    entity_data: T

class EntityDeletedEvent(DomainEvent):
    entity_id: UUID
Defining a custom event: 
from hexcore.domain.events import DomainEvent, EntityCreatedEvent
from uuid import UUID

class UserRegisteredEvent(EntityCreatedEvent[User]):
    """Fired when a new user completes registration."""
    pass

class PasswordChangedEvent(DomainEvent):
    user_id: UUID
    changed_by: UUID
Registering events on an entity and collecting them in a handler: 
# Inside a command handler after mutating the entity
user = User(username="alice", email="alice@example.com")
user.register_event(UserRegisteredEvent(entity_id=user.id, entity_data=user))

# The Unit of Work calls pull_domain_events() before committing
events = user.pull_domain_events()
for event in events:
    await event_bus.publish(event)
Domain events are immutable (frozen=True). Attempting to modify a field after construction raises a ValidationError. This is by design — events represent immutable facts.

​
Repositories

A repository abstracts the collection of domain objects, presenting them as an in-memory collection even when they are persisted in a database. Only aggregate roots have repositories. IBaseRepository[T] is the port (interface) that all repository implementations must fulfil: 
# hexcore/domain/repositories.py
class IBaseRepository(abc.ABC, Generic[T]):
    limit_offset_pagination: bool = True

    def __init__(self, uow: IUnitOfWork): ...

    @abstractmethod
    async def get_by_id(self, entity_id: UUID) -> T: ...

    async def get_active_by_id(self, entity_id: UUID) -> T:
        """Returns the entity only if is_active is True."""
        ...

    @abstractmethod
    async def list_all(
        self, limit: Optional[int] = None, offset: int = 0
    ) -> List[T]: ...

    @abstractmethod
    async def save(self, entity: T) -> T: ...

    @abstractmethod
    async def delete(self, entity: T) -> None: ...
Implementing an in-memory repository for tests: 
from uuid import UUID
from typing import Optional
from hexcore.domain.repositories import IBaseRepository
from hexcore.domain.uow import IUnitOfWork

class InMemoryUserRepository(IBaseRepository[User]):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)
        self._store: dict[UUID, User] = {}

    async def get_by_id(self, entity_id: UUID) -> User:
        user = self._store.get(entity_id)
        if user is None:
            raise UserNotFoundException(entity_id)
        return user

    async def list_all(
        self, limit: Optional[int] = None, offset: int = 0
    ) -> list[User]:
        items = list(self._store.values())[offset:]
        return items[:limit] if limit is not None else items

    async def save(self, entity: User) -> User:
        self._store[entity.id] = entity
        return entity

    async def delete(self, entity: User) -> None:
        self._store.pop(entity.id, None)

​
Domain Services

A domain service captures business logic that does not naturally belong to any single entity. HexCore’s BaseDomainService provides built-in filtering, sorting, searching, and pagination over entity collections without requiring any database: 
# hexcore/domain/services.py
class BaseDomainService:
    def __init__(self, event_bus: EventBus = ...) -> None:
        self.config = LazyConfig.get_config()
        self.event_bus = event_bus

    async def list_entities(
        self,
        repository: IBaseRepository[T],
        query: QueryRequestDTO,
    ) -> QueryResponseDTO: ...

    def query_entities(
        self,
        entities: Sequence[T],
        query: QueryRequestDTO,
    ) -> QueryResponseDTO: ...
Example domain service: 
from hexcore.domain.services import BaseDomainService
from hexcore.domain.repositories import IBaseRepository
from hexcore.application.dtos.query import QueryRequestDTO, QueryResponseDTO

class UserService(BaseDomainService):
    def __init__(self, user_repository: IBaseRepository[User]) -> None:
        super().__init__()
        self.user_repository = user_repository

    async def get_active_users(
        self, query: QueryRequestDTO
    ) -> QueryResponseDTO:
        return await self.list_entities(self.user_repository, query)

​
Unit of Work

The Unit of Work (UoW) manages transactions. It tracks which entities were mutated during a business operation, dispatches their domain events, and either commits or rolls back the entire unit atomically. IUnitOfWork is the port: 
# hexcore/domain/uow.py
class IUnitOfWork(abc.ABC):
    repositories: Dict[str, Any]
    event_bus: Any

    async def __aenter__(self) -> IUnitOfWork: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...

    @abstractmethod
    def collect_domain_events(self) -> List[Any]: ...

    @abstractmethod
    async def dispatch_events(self) -> None: ...

    @abstractmethod
    def clear_tracked_entities(self) -> None: ...

    @abstractmethod
    async def commit(self) -> None: ...

    @abstractmethod
    async def rollback(self) -> None: ...

    @abstractmethod
    def collect_entity(self, entity: BaseEntity) -> None: ...
Use it as an async context manager inside a command handler: 
class CreateUserHandler(AbstractCommandHandler[CreateUserCommand, UserDTO]):
    def __init__(self, uow: IUnitOfWork) -> None:
        self.uow = uow

    async def handle(self, command: CreateUserCommand) -> UserDTO:
        async with self.uow:
            repo: IBaseRepository[User] = self.uow.repositories["users"]
            user = User(username=command.username, email=command.email)

            # Track entity so the UoW can collect its domain events
            self.uow.collect_entity(user)
            user.register_event(
                UserRegisteredEvent(entity_id=user.id, entity_data=user)
            )

            await repo.save(user)
            await self.uow.commit()  # commits DB + dispatches events
            return UserDTO.model_validate(user)
IUnitOfWork.__aexit__ automatically calls rollback() if an exception escapes the async with block. You only need to call rollback() explicitly if you are handling partial failures within the block.

Build docs developers (and LLMs) love

Get started for freeTalk to us