Transactional Consistency with IUnitOfWork in HexCore

The Unit of Work (UoW) pattern groups one or more repository operations into a single atomic transaction. HexCore’s IUnitOfWork is an async context manager that opens a database transaction on entry, rolls back automatically on any exception, and provides a clean commit() / rollback() API for explicit control. It also acts as the event dispatcher — it tracks entities modified during the transaction and publishes their domain events after a successful commit. This page documents the interface, the get_repository() helper, and common usage patterns.

IUnitOfWork

IUnitOfWork lives in hexcore.domain.uow. It is an abstract base class; concrete implementations (e.g. for SQLAlchemy or Beanie) live in the infrastructure layer.

from hexcore.domain.uow import IUnitOfWork

Constructor

The base __init__ sets two instance attributes that concrete implementations extend:

repositories

dict[str, Any]

default:"{}"

Registry of repository instances attached to this UoW. Populated by the concrete implementation during __aenter__.

event_bus

EventBus | None

default:"None"

Reference to the active EventBus. Set by the concrete implementation. Used by dispatch_events() to publish collected domain events.

Async context manager

IUnitOfWork implements __aenter__ and __aexit__ so it can be used with async with:

async with uow:
    repo = get_repository(uow, "user_repository", IUserRepository)
    user = User.create(name="Alice", email="alice@example.com")
    await repo.save(user)
    await uow.commit()

If an exception propagates out of the async with block, __aexit__ automatically calls rollback() before re-raising the exception.

Abstract methods

commit()

Awaitable[None]

Flush all pending changes to the database and make them permanent. After a successful commit the concrete implementation should also call dispatch_events() to publish domain events from all tracked entities.

await uow.commit()

rollback()

Awaitable[None]

Discard all pending changes. Called automatically by __aexit__ when an exception occurs. You can also call it explicitly to abort a transaction.

await uow.rollback()

collect_entity(entity)

None

Register a BaseEntity instance for event collection. The UoW will call entity.pull_domain_events() during collect_domain_events() to gather events before dispatching them.

uow.collect_entity(user)

You rarely call collect_entity directly. The @register_entity_on_uow decorator (applied to repository save() implementations) calls this automatically whenever an entity has pending domain events.

collect_domain_events()

List[DomainEvent]

Pull domain events from all tracked entities and return them as a flat list. Clears the entity’s event queue in the process (via pull_domain_events()).

events = uow.collect_domain_events()

dispatch_events()

Awaitable[None]

Collect domain events from all tracked entities and publish each one to the event_bus. Typically called inside commit() after the database transaction succeeds.

await uow.dispatch_events()

clear_tracked_entities()

None

Remove all entity references from the UoW’s tracking list without pulling or dispatching events. Useful for resetting state between test cases.

uow.clear_tracked_entities()

@register_entity_on_uow decorator

hexcore.infrastructure.uow.decorators provides the @register_entity_on_uow decorator. Apply it to a repository’s save() method to make the UoW automatically track any entity that has pending domain events.

from hexcore.infrastructure.uow.decorators import register_entity_on_uow

class SQLAlchemyUserRepository(IUserRepository):

    @register_entity_on_uow
    async def save(self, entity: User) -> User:
        # ... persist to database ...
        return entity

Under the hood the decorator checks entity._domain_events after the wrapped method returns. If the list is non-empty it calls self.uow.collect_entity(entity).

get_repository() helper

hexcore.infrastructure.uow.helpers.get_repository provides type-safe access to a named repository on a UoW instance.

from hexcore.infrastructure.uow.helpers import get_repository

def get_repository(uow: IUnitOfWork, repo_name: str, repo_type: Type[T]) -> T

uow

IUnitOfWork

The active Unit of Work instance.

repo_name

str

The attribute name under which the repository is attached to the UoW (e.g. "user_repository").

repo_type

Type[T]

The repository interface or class used for type casting. Not used for runtime lookup — only for static type checking.

If the attribute does not exist on the UoW, get_repository raises an AttributeError with a helpful message listing every IBaseRepository instance it found, making debugging easy:

AttributeError: El repositorio 'order_repository' (IOrderRepository) no esta disponible en
SQLAlchemyUoW. Repositorios detectados: product_repository, user_repository.

Usage patterns

Basic transaction
Explicit rollback
Event dispatch

The most common pattern — open a UoW, perform operations, commit.

from hexcore.infrastructure.uow.helpers import get_repository
from src.domain.users.repositories import IUserRepository
from src.domain.users.entities import User


async def register_user(name: str, email: str, uow: IUnitOfWork) -> User:
    async with uow:
        repo = get_repository(uow, "user_repository", IUserRepository)
        user = User.create(name=name, email=email)
        saved = await repo.save(user)
        await uow.commit()
        return saved

Roll back manually when business rules are violated inside a transaction.

async def transfer_credits(
    from_id: UUID,
    to_id: UUID,
    amount: int,
    uow: IUnitOfWork,
) -> None:
    async with uow:
        repo = get_repository(uow, "account_repository", IAccountRepository)
        source = await repo.get_active_by_id(from_id)
        target = await repo.get_active_by_id(to_id)

        if source.credits < amount:
            await uow.rollback()
            raise InsufficientCreditsError(source.id)

        source.deduct(amount)
        target.credit(amount)
        await repo.save(source)
        await repo.save(target)
        await uow.commit()

Domain events registered on entities are published automatically during commit(). The example below shows the flow explicitly.

async def promote_user(user_id: UUID, uow: IUnitOfWork) -> None:
    async with uow:
        repo = get_repository(uow, "user_repository", IUserRepository)
        user = await repo.get_active_by_id(user_id)

        # Entity registers a UserPromotedEvent internally
        user.promote_to_admin()

        # save() calls @register_entity_on_uow → uow.collect_entity(user)
        await repo.save(user)

        # commit() → dispatch_events() → event_bus.publish(UserPromotedEvent)
        await uow.commit()

Repository discovery and v2 behaviour

v2 breaking change. If repository_discovery_paths in your ServerConfig is empty (the default), the Unit of Work cannot discover any repository implementations. Any attempt to access a repository via get_repository() will fail with an explicit AttributeError. Always populate repository_discovery_paths in your config.py.

config.py

from hexcore.config import ServerConfig

config = ServerConfig(
    repository_discovery_paths={
        "src.infrastructure.repositories",
    }
)

The UoW scans every module listed in repository_discovery_paths at startup and instantiates the repository classes it finds, attaching them to the UoW instance by a conventional attribute name. If a module is missing from the list, its repositories are invisible to get_repository().

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Transactional Consistency with IUnitOfWork in HexCore

IUnitOfWork

Constructor

Async context manager

Abstract methods

@register_entity_on_uow decorator

get_repository() helper

Usage patterns

Repository discovery and v2 behaviour

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Documentation Index

​IUnitOfWork

​Constructor

​Async context manager

​Abstract methods

​@register_entity_on_uow decorator

​get_repository() helper

​Usage patterns

​Repository discovery and v2 behaviour

Build docs developers (and LLMs) love

IUnitOfWork

Constructor

Async context manager

Abstract methods

@register_entity_on_uow decorator

get_repository() helper

Usage patterns

Repository discovery and v2 behaviour