SqlAlchemyRepository — SQLAlchemy CRUD Repository

SqlAlchemyRepository is HexCore’s built-in async CRUD base for SQLAlchemy. It wires a typed domain entity T to a SQLAlchemy ORM model M, handling serialisation, deserialisation, and field-level transformation through optional resolver and serialiser maps. You extend it by implementing a handful of abstract properties—HexCore does the rest.

SqlAlchemyRepository (and all symbols in this page) is only available when the hexcore[sql] extra is installed:

pip install "hexcore[sql]"

If SQLAlchemy is not installed the class silently becomes an empty generic stub so that imports in non-SQL services never fail.

Class hierarchy

IBaseRepository[T]
  └── BaseSQLAlchemyRepository[T]           # session injection
        └── SqlAlchemyRepository[T, M]      # CRUD implementations

SqlAlchemyRepository is also exported under the backward-compatible alias SQLAlchemyCommonImplementationsRepo.

`BaseSQLAlchemyRepository`

hexcore.infrastructure.repositories.base.BaseSQLAlchemyRepository[T] The abstract base that injects the SQLAlchemy AsyncSession from a IUnitOfWork object at construction time. All concrete SQL repositories must ultimately derive from this class.

Constructor

BaseSQLAlchemyRepository(uow: IUnitOfWork)

uow

IUnitOfWork

required

A unit-of-work instance that exposes a session attribute of type AsyncSession. HexCore’s built-in UoW implementations satisfy this contract automatically.

Properties

session

AsyncSession

The AsyncSession extracted from the UoW at construction time. Raises ValueError if the UoW did not provide a session.

`BaseModel`

hexcore.infrastructure.repositories.orms.sqlalchemy.BaseModel[T] Abstract SQLAlchemy declarative base for all ORM models in a HexCore project. Every table-mapped class must inherit from BaseModel (or from Base directly if you need a model without the standard audit columns).

from hexcore.infrastructure.repositories.orms.sqlalchemy import BaseModel
from hexcore.domain.base import BaseEntity

class UserModel(BaseModel["UserEntity"]):
    __tablename__ = "users"

    name: Mapped[str] = mapped_column(String(255))
    email: Mapped[str] = mapped_column(String(255), unique=True)

Built-in columns

Column	Type	Default
`id`	`UUID` (primary key)	`uuid4()`
`is_active`	`Boolean`	`True`
`created_at`	`DateTime(timezone=True)`	`datetime.now(UTC)`
`updated_at`	`DateTime(timezone=True)`	`datetime.now(UTC)`, auto-updated on `UPDATE`

Methods

set_domain_entity(entity: T)

None

Stores the associated domain entity on the model instance. Called automatically by save_entity.

get_domain_entity()

Returns the previously stored domain entity.

`SqlAlchemyRepository`

hexcore.infrastructure.repositories.implementations.SqlAlchemyRepository[T, M] Full CRUD repository that maps a domain entity T to an ORM model M. All methods are async.

Generic parameters

Parameter	Bound	Description
`T`	`BaseEntity`	Domain entity type returned by every method
`M`	`BaseModel[Any]`	SQLAlchemy model type used for DB persistence

Abstract properties to implement

entity_cls

Type[T]

required

The domain entity class. Used to reconstruct entities from ORM rows via model_construct.

model_cls

Type[M]

required

The SQLAlchemy model class. Used to build SELECT, INSERT, and MERGE statements.

not_found_exception

Type[Exception]

required

Exception class (not instance) raised when a record is not found. Receives the queried entity_id as its sole constructor argument.

Optional properties

fields_resolvers

FieldResolversType[M]

default:"{}"

A mapping of { entity_field: (model_field, async_callable) } applied when converting a model row to a domain entity. Each async callable receives the full model instance and must return the resolved value. Useful for loading related entities or computing derived fields.

fields_serializers

FieldSerializersType[T]

default:"{}"

A mapping of { entity_field: (model_field, callable) } applied when converting a domain entity to an ORM model. Useful for serialising value objects or embedded types to flat columns.

limit_offset_pagination

bool

default:"True"

When True (the default, inherited from IBaseRepository), list_all passes limit and offset to the database query. Set to False to disable pagination for this repository.

Methods

get_by_id(entity_id: UUID)

Awaitable[T]

Fetches a single record by primary key. Raises not_found_exception(entity_id) if no row is found.

list_all(limit: Optional[int] = None, offset: int = 0)

Awaitable[List[T]]

Returns all active records. When limit_offset_pagination is True, applies OFFSET offset LIMIT limit. Relationships are eagerly loaded via selectinload.

query_all(query: QueryRequestDTO)

Awaitable[tuple[List[T], int]]

Executes a parameterised query supporting full-text search, typed filters (EQ, NE, GT, GTE, LT, LTE, IN, NOT_IN, CONTAINS, STARTSWITH, ENDSWITH, IS_NULL), sorting, and limit/offset pagination. Returns the page of results and the total matching row count.

save(entity: T)

Awaitable[T]

Persists a domain entity using SQLAlchemy’s session.merge() (upsert semantics), flushes, refreshes, and returns the re-hydrated entity.

delete(entity: T)

Awaitable[None]

Performs a logical delete: calls entity.deactivate(), then saves the updated entity. Rows are never physically removed.

Session utilities

hexcore.infrastructure.repositories.orms.sqlalchemy.session

Function	Signature	Description
`get_engine()`	`() -> AsyncEngine`	Returns the lazy-initialised async engine built from `ServerConfig.async_sql_database_url`.
`get_session_factory()`	`() -> async_sessionmaker[AsyncSession]`	Returns the lazy-initialised session factory.
`get_async_db_session()`	`async () -> AsyncGenerator[AsyncSession, None]`	Async generator yielding a single session; suitable for FastAPI’s `Depends`.
`reset_sqlalchemy_engine()`	`async () -> None`	Disposes the current engine and clears the factory. Required when forking processes (e.g. Celery workers).

Complete example

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
from hexcore.infrastructure.repositories.orms.sqlalchemy import BaseModel
from myapp.domain.entities import UserEntity

class UserModel(BaseModel["UserEntity"]):
    __tablename__ = "users"

    name: Mapped[str] = mapped_column(String(255))
    email: Mapped[str] = mapped_column(String(255), unique=True)

Repository discovery

HexCore can auto-discover all BaseSQLAlchemyRepository subclasses in your project. Register your repository packages in ServerConfig:

config.py

from hexcore.config import ServerConfig

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

Then call:

from hexcore.infrastructure.repositories.utils import discover_sql_repositories

repos = discover_sql_repositories()
# {"user": UserRepository, "product": ProductRepository, ...}

Repository keys are derived by stripping the Repository / Repo suffix and lowercasing the remaining class name. A class named UserRepository yields key "user".

Domain

CQRS

Infrastructure

SqlAlchemyRepository — SQLAlchemy CRUD Repository

Class hierarchy

`BaseSQLAlchemyRepository`

Constructor

Properties

`BaseModel`

Built-in columns

Methods

`SqlAlchemyRepository`

Generic parameters

Abstract properties to implement

Optional properties

Methods

Session utilities

Complete example

Repository discovery

Build docs developers (and LLMs) love

Domain

CQRS

Infrastructure

Documentation Index

​Class hierarchy

​BaseSQLAlchemyRepository

​Constructor

​Properties

​BaseModel

​Built-in columns

​Methods

​SqlAlchemyRepository

​Generic parameters

​Abstract properties to implement

​Optional properties

​Methods

​Session utilities

​Complete example

​Repository discovery

Build docs developers (and LLMs) love

Class hierarchy

`BaseSQLAlchemyRepository`

Constructor

Properties

`BaseModel`

Built-in columns

Methods

`SqlAlchemyRepository`

Generic parameters

Abstract properties to implement

Optional properties

Methods

Session utilities

Complete example

Repository discovery