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.

IUnitOfWork defines the transactional boundary for every write operation in your application. It acts as an async context manager: entering the context opens a transaction, a successful exit commits it, and any unhandled exception automatically triggers a rollback. The UoW also owns the event-dispatch lifecycle — it collects domain events from tracked entities and publishes them after a successful commit. Concrete implementations live in the infrastructure layer (e.g., SQLAlchemy, Django ORM). 
from hexcore.domain.uow import IUnitOfWork

​
Class definition

class IUnitOfWork(abc.ABC):
    def __init__(self) -> None:
        self.repositories: Dict[str, Any] = {}
        self.event_bus: Any = None

​
Constructor attributes

​
repositories
Dict[str, Any]
default:"{}"
A mapping of repository names to repository instances. Populated by the concrete implementation’s __init__. Used by get_repository() for type-safe access.
​
event_bus
Any
default:"None"
An optional EventBus (or IEventBus) instance. When set, dispatch_events() uses it to publish collected domain events.

​
Async context manager

​
__aenter__

async def __aenter__(self) -> IUnitOfWork
Enters the transactional context and returns self. Override in your concrete implementation to open a database session or begin a transaction.
​
uow
IUnitOfWork
The same Unit of Work instance, ready for use inside the async with block.

​
__aexit__

async def __aexit__(
    self,
    exc_type: Optional[type],
    exc_val: Optional[BaseException],
    exc_tb: Optional[Any],
) -> None
Exits the transactional context. If exc_type is set (an exception was raised inside the block), rollback() is called automatically. Successful exits do not commit automatically — you must call commit() explicitly before the async with block ends.
The design choice of requiring an explicit commit() rather than auto-committing on success makes the commit intent visible in application code and prevents accidental commits when control flow is complex.

​
Abstract methods

​
commit

@abstractmethod
async def commit(self) -> None
Persists all pending changes to the data store within the current transaction. Call this once all repository mutations and event registrations are complete, before the async with block exits.

​
rollback

@abstractmethod
async def rollback(self) -> None
Aborts the current transaction and discards all pending changes. Called automatically by __aexit__ on exception; you may also call it manually inside the async with block to programmatically abort.

​
collect_entity

@abstractmethod
def collect_entity(self, entity: BaseEntity) -> None
Registers a BaseEntity for event tracking. Concrete implementations typically store the entity in an internal set and later call entity.pull_domain_events() inside collect_domain_events().
​
entity
BaseEntity
required
The domain entity whose events should be harvested on commit.

​
collect_domain_events

@abstractmethod
def collect_domain_events(self) -> List[Any]
Iterates over all tracked entities, calls pull_domain_events() on each, and returns a flat list of all accumulated DomainEvent instances.
​
events
List[DomainEvent]
All domain events collected from tracked entities since the last call.

​
dispatch_events

@abstractmethod
async def dispatch_events(self) -> None
Publishes all collected domain events through event_bus. Should be called after a successful commit() so events are only published when the transaction has been durably persisted.

​
clear_tracked_entities

@abstractmethod
def clear_tracked_entities(self) -> None
Clears the internal set of tracked entities. Typically called at the end of dispatch_events() or inside rollback() to ensure no stale entity references are held between transactions.

​
get_repository helper

from hexcore.infrastructure.uow.helpers import get_repository

def get_repository(uow: IUnitOfWork, repo_name: str, repo_type: Type[T]) -> T
A type-safe accessor for repositories stored on a UoW instance. Instead of writing uow.users directly (which loses type information), get_repository casts the attribute to the expected repository type and provides a helpful error message if the attribute does not exist.
​
uow
IUnitOfWork
required
The Unit of Work instance to inspect.
​
repo_name
str
required
The attribute name of the repository on the UoW (e.g., "users", "products").
​
repo_type
Type[T]
required
The expected repository class. Used for the cast and the error message only — no runtime isinstance check is performed.
​
repository
T
The repository instance cast to repo_type.
Raises: AttributeError — if repo_name is not an attribute on uow. The error message lists all repositories detected on the UoW instance (attributes whose values are IBaseRepository instances), making debugging straightforward.

​
Usage example

from sqlalchemy.ext.asyncio import AsyncSession
from hexcore.domain.uow import IUnitOfWork
from hexcore.domain.base import BaseEntity
from myapp.repositories import UserRepository, OrderRepository

class SQLAlchemyUnitOfWork(IUnitOfWork):
    def __init__(self, session_factory, event_bus=None):
        super().__init__()
        self._session_factory = session_factory
        self.event_bus = event_bus
        self._tracked: set[BaseEntity] = set()

    async def __aenter__(self):
        self.session: AsyncSession = self._session_factory()
        self.users = UserRepository(self)
        self.orders = OrderRepository(self)
        self.repositories = {"users": self.users, "orders": self.orders}
        return await super().__aenter__()

    async def commit(self):
        await self.session.commit()

    async def rollback(self):
        await self.session.rollback()

    def collect_entity(self, entity: BaseEntity) -> None:
        self._tracked.add(entity)

    def collect_domain_events(self):
        events = []
        for entity in self._tracked:
            events.extend(entity.pull_domain_events())
        return events

    async def dispatch_events(self):
        if self.event_bus:
            for event in self.collect_domain_events():
                await self.event_bus.publish(event)
        self.clear_tracked_entities()

    def clear_tracked_entities(self):
        self._tracked.clear()
Always call commit() before dispatch_events(). Publishing events before the transaction is durable risks notifying other systems about changes that may still be rolled back.

Build docs developers (and LLMs) love

Get started for freeTalk to us