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.

IBaseRepository is the abstract contract every repository adapter must implement. It is generic over T (any BaseEntity subclass), so a concrete SQLAlchemyUserRepository can declare IBaseRepository[User] and the entire CRUD surface becomes type-safe. The interface ships with one concrete helper method, get_active_by_id, that adds a soft-delete guard on top of the abstract get_by_id without requiring you to duplicate that logic in every adapter.
from hexcore.domain.repositories import IBaseRepository

Class definition

class IBaseRepository(abc.ABC, Generic[T]):
    limit_offset_pagination: bool = True

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

Class attribute

limit_offset_pagination
bool
default:"True"
Signals to callers (and documentation) that this repository implementation honours the limit / offset parameters of list_all. Override to False in adapters that use cursor-based pagination or return full result sets.

Constructor

uow
IUnitOfWork
required
The Unit of Work instance this repository belongs to. Stored on self.uow and used to participate in the same transactional boundary as other repositories registered with the same UoW.

Abstract methods

The following four methods must be implemented by every concrete repository. All are async.

get_by_id

@abstractmethod
async def get_by_id(self, entity_id: UUID) -> T
Fetches a single entity by its primary key. Raise an appropriate domain or infrastructure exception when the entity is not found (e.g., EntityNotFoundException).
entity_id
UUID
required
The UUID primary key of the entity to retrieve.
entity
T
The found entity instance, including inactive ones. Use get_active_by_id if you need the soft-delete guard.

list_all

@abstractmethod
async def list_all(
    self,
    limit: Optional[int] = None,
    offset: int = 0,
) -> List[T]
Returns a paginated (or full) list of entities. When limit_offset_pagination is True, implementations are expected to apply limit and offset to the underlying query.
limit
int | None
default:"None"
Maximum number of entities to return. None means no limit.
offset
int
default:"0"
Number of entities to skip before collecting results.
entities
List[T]
The (possibly paginated) list of entity instances.

save

@abstractmethod
async def save(self, entity: T) -> T
Persists a new or updated entity. Implementations typically upsert (insert or update) the record and return the persisted state, which may differ from the input when the data store adds computed fields (e.g., database-level updated_at).
entity
T
required
The entity instance to create or update.
entity
T
The entity as it exists in the data store after the save operation.

delete

@abstractmethod
async def delete(self, entity: T) -> None
Removes the entity from the data store. Adapters may implement this as a hard DELETE or delegate to entity.deactivate() for logical deletion, depending on project requirements.
entity
T
required
The entity instance to remove.

Concrete method

get_active_by_id

async def get_active_by_id(self, entity_id: UUID) -> T
Calls get_by_id and then checks entity.is_active. Raises InactiveEntityException if the entity exists but is logically deleted. This guard is pre-built so you do not need to repeat it in every adapter.
entity_id
UUID
required
The UUID of the entity to retrieve. The entity must exist and have is_active=True.
entity
T
The active entity instance.
Raises: InactiveEntityException — if entity.is_active is False.

InactiveEntityException

from hexcore.domain.exceptions import InactiveEntityException
class InactiveEntityException(Exception):
    def __init__(self) -> None:
        super().__init__("La entidad no está activa.")
Raised by get_active_by_id when a requested entity is found in the data store but has been soft-deleted (is_active=False). Catch this in your application layer to return a 404 or 410 HTTP response rather than exposing inactive data.
InactiveEntityException carries no entity metadata. If you need the entity ID in error handling code, fetch it from the exception context or re-wrap the exception in your own application-layer error type.

Usage example

from uuid import UUID
from typing import Optional, List
from hexcore.domain.repositories import IBaseRepository
from myapp.domain.entities import Product

class SQLAlchemyProductRepository(IBaseRepository[Product]):
    async def get_by_id(self, entity_id: UUID) -> Product:
        row = await self.uow.session.get(Product, entity_id)
        if row is None:
            raise ValueError(f"Product {entity_id} not found")
        return row

    async def list_all(
        self,
        limit: Optional[int] = None,
        offset: int = 0,
    ) -> List[Product]:
        query = select(Product).offset(offset)
        if limit is not None:
            query = query.limit(limit)
        result = await self.uow.session.execute(query)
        return list(result.scalars())

    async def save(self, entity: Product) -> Product:
        self.uow.session.add(entity)
        return entity

    async def delete(self, entity: Product) -> None:
        await self.uow.session.delete(entity)

Build docs developers (and LLMs) love