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.

Repositories are the bridge between your domain model and the database. In HexCore every repository starts as an abstract interface in the domain layer — a subclass of IBaseRepository[T] — and is fulfilled by a concrete implementation in the infrastructure layer. This separation ensures your domain stays free of database concerns and remains easily testable. This page covers the IBaseRepository contract, the limit_offset_pagination flag, and a full end-to-end example.

​
IBaseRepository[T]

IBaseRepository lives in hexcore.domain.repositories. It is a generic, abstract base class that defines the minimum CRUD contract every repository must honour. 
from hexcore.domain.repositories import IBaseRepository

​
Constructor

​
uow
IUnitOfWork
The Unit of Work that owns this repository instance. Injected automatically when the UoW is initialised. The repository stores it as self.uow for use in transaction-aware operations.

​
Class attribute

​
limit_offset_pagination
bool
default:"True"
When True (the default), list_all() implementations should respect the limit and offset arguments and apply pagination at the database level. Set this to False if the underlying data source handles pagination differently or not at all.

​
Abstract methods

​
get_by_id(entity_id)
Awaitable[T]
Fetch a single entity by its UUID primary key. Raise a domain-specific NotFoundException (or equivalent) if no record is found.
async def get_by_id(self, entity_id: UUID) -> T:
    raise NotImplementedError
​
list_all(limit, offset)
Awaitable[List[T]]
Return a list of entities. Both parameters are optional.
ParameterTypeDefaultDescription
limitint | NoneNoneMaximum number of records to return. None means no limit.
offsetint0Number of records to skip.
async def list_all(
    self, limit: int | None = None, offset: int = 0
) -> list[T]:
    raise NotImplementedError
​
save(entity)
Awaitable[T]
Persist a new or modified entity. Returns the saved entity (which may have database-generated fields updated, such as an auto-increment ID or server-side timestamp).
async def save(self, entity: T) -> T:
    raise NotImplementedError
​
delete(entity)
Awaitable[None]
Remove an entity from the store. Whether this performs a hard or soft delete depends on the concrete implementation.
async def delete(self, entity: T) -> None:
    raise NotImplementedError

​
Concrete method

​
get_active_by_id(entity_id)
Awaitable[T]
Calls get_by_id() and then checks entity.is_active. If the entity is inactive, raises InactiveEntityException. Use this method instead of get_by_id() when you need to enforce the soft-delete contract.
# Provided by IBaseRepository — no need to override
user = await repo.get_active_by_id(user_id)

​
Defining a domain interface

Create a subclass of IBaseRepository in your domain layer. Keep it purely abstract — no database imports. 
# src/domain/users/repositories.py
from __future__ import annotations
import abc
from uuid import UUID
from hexcore.domain.repositories import IBaseRepository
from .entities import User


class IUserRepository(IBaseRepository[User]):
    """
    Domain contract for User persistence.
    The infrastructure layer provides the concrete implementation.
    """

    @abc.abstractmethod
    async def get_by_email(self, email: str) -> User:
        """Find a user by their unique email address."""
        raise NotImplementedError
Add only domain-meaningful query methods here — get_by_email, find_active_admins, etc. Resist the urge to add generic filter methods; those belong in the service layer via BaseDomainService.list_entities().

​
Implementing in infrastructure

The infrastructure layer provides the concrete class that connects the domain interface to the database. The example below shows a SQLAlchemy implementation stub that extends IUserRepository. 
# src/infrastructure/repositories/user_repository.py
from __future__ import annotations
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from hexcore.domain.uow import IUnitOfWork
from src.domain.users.entities import User
from src.domain.users.repositories import IUserRepository
from src.infrastructure.database.models.user_model import UserModel


class UserNotFoundException(Exception):
    pass


class SQLAlchemyUserRepository(IUserRepository):
    """Concrete SQLAlchemy implementation of IUserRepository."""

    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def session(self) -> AsyncSession:
        # Retrieve the async session stored on the UoW
        return self.uow.session  # type: ignore[attr-defined]

    async def get_by_id(self, entity_id: UUID) -> User:
        result = await self.session.get(UserModel, entity_id)
        if result is None:
            raise UserNotFoundException(f"User {entity_id} not found")
        return User.model_validate(result)

    async def get_by_email(self, email: str) -> User:
        stmt = select(UserModel).where(UserModel.email == email)
        row = (await self.session.execute(stmt)).scalar_one_or_none()
        if row is None:
            raise UserNotFoundException(f"User with email {email!r} not found")
        return User.model_validate(row)

    async def list_all(
        self, limit: int | None = None, offset: int = 0
    ) -> list[User]:
        stmt = select(UserModel).offset(offset)
        if limit is not None:
            stmt = stmt.limit(limit)
        rows = (await self.session.execute(stmt)).scalars().all()
        return [User.model_validate(row) for row in rows]

    async def save(self, entity: User) -> User:
        model = UserModel(**entity.model_dump())
        self.session.add(model)
        await self.session.flush()
        return User.model_validate(model)

    async def delete(self, entity: User) -> None:
        model = await self.session.get(UserModel, entity.id)
        if model is not None:
            await self.session.delete(model)

​
Registering repositories for discovery

For HexCore’s Unit of Work to find and wire your repository, the module containing the concrete implementation must be listed in repository_discovery_paths inside your config.py:
config.py
from hexcore.config import ServerConfig

config = ServerConfig(
    repository_discovery_paths={
        "src.infrastructure.repositories",
    }
)
If a repository’s module is not listed in repository_discovery_paths, the Unit of Work will not load it and will raise an explicit error when you try to access it. This is a deliberate v2 breaking change — there is no implicit auto-discovery fallback.

​
Using a repository in application code

Once registered, repositories are accessed through the Unit of Work: 
from hexcore.infrastructure.uow.helpers import get_repository
from src.domain.users.repositories import IUserRepository


async def create_user(name: str, email: str, uow) -> dict:
    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 {"id": str(saved.id), "email": saved.email}
See the Unit of Work guide for the full transaction lifecycle.

Build docs developers (and LLMs) love

Get started for freeTalk to us