Async SQLAlchemy Repositories with HexCore Framework

HexCore’s SqlAlchemyRepository (also exported as SQLAlchemyCommonImplementationsRepo) is a generic base class that wires your SQLAlchemy async models to domain entities. This guide covers defining your ORM model, implementing a concrete repository, configuring field resolvers and serializers for complex types, and running migrations — all the way from an empty table to a working end-to-end flow.

Prerequisites

Install HexCore with the SQLAlchemy extra:

pip install "hexcore[sqlalchemy]"

Make sure async_sql_database_url is set in your ServerConfig:

config.py

from hexcore.config import ServerConfig

config = ServerConfig(
    async_sql_database_url="postgresql+asyncpg://user:pass@localhost/mydb",
    sql_database_url="postgresql://user:pass@localhost/mydb",  # used by Alembic
    repository_discovery_paths={
        "src.infrastructure.repositories",
    },
)

Core Classes

BaseModel

Abstract SQLAlchemy declarative base. All ORM models must subclass it. Provides id (UUID PK), is_active (bool), created_at, and updated_at columns automatically.

SqlAlchemyRepository

Generic CRUD repository. Subclass it with your entity type T and model type M. Provides get_by_id, list_all, query_all, save, and delete.

Step 1 — Define the ORM Model

Import BaseModel and Base from HexCore’s SQLAlchemy ORM module. Your model extends BaseModel[YourEntity] and declares the __tablename__ plus any extra columns.

src/infrastructure/database/models/user_model.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from hexcore.infrastructure.repositories.orms.sqlalchemy import Base, BaseModel
from src.domain.users.entities import User


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

    name: Mapped[str] = mapped_column(String(255), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    role: Mapped[str] = mapped_column(String(50), nullable=False, default="member")

BaseModel already defines id (UUID primary key), is_active, created_at, and updated_at. Do not redeclare them in your subclass.

Step 2 — Define the Domain Entity

src/domain/users/entities.py

from uuid import UUID
from hexcore.domain.base import BaseEntity


class User(BaseEntity):
    name: str
    email: str
    role: str

Step 3 — Define the Repository

Subclass SqlAlchemyRepository[Entity, Model] and declare the four required properties: entity_cls, model_cls, not_found_exception, and uow (received via constructor).

src/infrastructure/repositories/user_repository.py

from hexcore.infrastructure.repositories.implementations import SqlAlchemyRepository
from hexcore.domain.uow import IUnitOfWork

from src.domain.users.entities import User
from src.domain.users.exceptions import UserNotFoundException
from src.domain.users.repositories import IUserRepository
from src.infrastructure.database.models.user_model import UserModel


class UserRepository(SqlAlchemyRepository[User, UserModel], IUserRepository):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return User

    @property
    def model_cls(self):
        return UserModel

    @property
    def not_found_exception(self):
        return UserNotFoundException

That’s it. The repository inherits get_by_id, list_all, query_all, save, and delete for free.

Available Methods

get_by_id

async (entity_id: UUID) -> T

Fetches a single record by its primary key UUID. Raises not_found_exception if missing.

list_all

async (limit: int | None, offset: int) -> list[T]

Returns all records. limit_offset_pagination defaults to True, so limit and offset are applied automatically. Set it to False on the repository class to always return every row.

query_all

async (query: QueryRequestDTO) -> tuple[list[T], int]

Executes a filtered, sorted, paginated query. Returns a (items, total_count) tuple. See QueryRequestDTO for filter operators and sort directions.

save

async (entity: T) -> T

Upserts the entity (insert or update via session.merge). Returns the refreshed entity.

delete

async (entity: T) -> None

Performs a logical delete — sets is_active = False rather than issuing a SQL DELETE.

Field Resolvers and Serializers

When a column stores a raw primitive but your entity field is a rich type (e.g. an Enum, a nested entity, or a value object), use fields_resolvers and fields_serializers to bridge the gap.

`fields_resolvers` — Model → Entity

Resolvers run after the model row is loaded and translate raw column values into entity-friendly types. Each entry maps an entity field name to a (source_column, async_resolver_fn) tuple. The resolver receives the full model instance.

src/infrastructure/repositories/order_repository.py

from hexcore.infrastructure.repositories.decorators import cycle_protection_resolver
from hexcore.infrastructure.repositories.implementations import SqlAlchemyRepository
from hexcore.domain.uow import IUnitOfWork

from src.domain.orders.entities import Order
from src.domain.orders.enums import OrderStatus
from src.domain.orders.exceptions import OrderNotFoundException
from src.infrastructure.database.models.order_model import OrderModel


class OrderRepository(SqlAlchemyRepository[Order, OrderModel]):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return Order

    @property
    def model_cls(self):
        return OrderModel

    @property
    def not_found_exception(self):
        return OrderNotFoundException

    @property
    def fields_resolvers(self):
        @cycle_protection_resolver
        async def resolve_status(model: OrderModel):
            return OrderStatus(model.status)

        return {
            # entity_field: (model_column, resolver_fn)
            "status": ("status", resolve_status),
        }

`fields_serializers` — Entity → Model

Serializers run before the entity is persisted and convert complex entity fields into column-compatible primitives. Each entry maps an entity field to a (dest_column, sync_serializer_fn) tuple. The serializer receives the full entity.

    @property
    def fields_serializers(self):
        return {
            # entity_field: (model_column, serializer_fn)
            "status": ("status", lambda entity: entity.status.value),
        }

Use the @cycle_protection_resolver decorator on resolver functions that navigate relationships. It automatically breaks circular reference chains by caching already-visited entity IDs in a context variable.

Enabling Limit/Offset Pagination

limit_offset_pagination defaults to True, so list_all respects limit and offset out of the box. To disable pagination and always return every row, set it to False in the repository class body:

class UserRepository(SqlAlchemyRepository[User, UserModel], IUserRepository):
    limit_offset_pagination = False
    ...

Session Factory

HexCore manages the async SQLAlchemy engine and session factory lazily. You can access them directly if needed — for example, in FastAPI dependency injection:

src/infrastructure/database/session.py

from hexcore.infrastructure.repositories.orms.sqlalchemy.session import (
    get_async_db_session,
    get_engine,
    get_session_factory,
    reset_sqlalchemy_engine,
)

# Use as a FastAPI dependency:
# async def get_db():
#     async for session in get_async_db_session():
#         yield session

In long-running worker processes (RabbitMQ, Celery), call await reset_sqlalchemy_engine() at startup to rebind the connection pool to the current asyncio event loop and avoid “Task attached to a different loop” errors.

Registering Models for Alembic

Use import_all_models from HexCore’s SQLAlchemy utilities to ensure Alembic discovers every model before generating migrations:

alembic/env.py

from alembic import context
from hexcore.config import LazyConfig
from hexcore.infrastructure.repositories.orms.sqlalchemy import Base
from hexcore.infrastructure.repositories.orms.sqlalchemy.utils import import_all_models
import src.infrastructure.database.models as models

# Auto-import every model module so Alembic sees all metadata
import_all_models(models)

# Tell Alembic which metadata to diff against
target_metadata = Base.metadata

# Pull the sync DB URL from config
config = context.config
database_url = LazyConfig.get_config().sql_database_url
if database_url:
    config.set_main_option("sqlalchemy.url", database_url)

Complete End-to-End Example

Define the entity and exception

src/domain/users/entities.py

from uuid import UUID
from hexcore.domain.base import BaseEntity

class User(BaseEntity):
    name: str
    email: str
    role: str

src/domain/users/exceptions.py

class UserNotFoundException(Exception):
    def __init__(self, user_id):
        super().__init__(f"User {user_id} not found.")

Define the ORM model

src/infrastructure/database/models/user_model.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from hexcore.infrastructure.repositories.orms.sqlalchemy import BaseModel
from src.domain.users.entities import User

class UserModel(BaseModel[User]):
    __tablename__ = "users"
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    role: Mapped[str] = mapped_column(String(50), default="member")

Implement the repository

src/infrastructure/repositories/user_repository.py

from hexcore.infrastructure.repositories.implementations import SqlAlchemyRepository
from hexcore.domain.uow import IUnitOfWork
from src.domain.users.entities import User
from src.domain.users.exceptions import UserNotFoundException
from src.infrastructure.database.models.user_model import UserModel

class UserRepository(SqlAlchemyRepository[User, UserModel]):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return User

    @property
    def model_cls(self):
        return UserModel

    @property
    def not_found_exception(self):
        return UserNotFoundException

Generate and run migrations

hexcore make-migrations "add users table"
hexcore migrate

Use the repository in a use case

src/application/use_cases/create_user.py

from uuid import uuid4
from src.domain.users.entities import User
from src.domain.users.repositories import IUserRepository
from hexcore.domain.uow import IUnitOfWork

class CreateUserUseCase:
    def __init__(self, repo: IUserRepository, uow: IUnitOfWork) -> None:
        self.repo = repo
        self.uow = uow

    async def execute(self, name: str, email: str) -> User:
        user = User(id=uuid4(), name=name, email=email, role="member")
        async with self.uow:
            saved = await self.repo.save(user)
            await self.uow.commit()
        return saved

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Async SQLAlchemy Repositories with HexCore Framework

Prerequisites

Core Classes

BaseModel

SqlAlchemyRepository

Step 1 — Define the ORM Model

Step 2 — Define the Domain Entity

Step 3 — Define the Repository

Available Methods

Field Resolvers and Serializers

`fields_resolvers` — Model → Entity

`fields_serializers` — Entity → Model

Session Factory

Registering Models for Alembic

Complete End-to-End Example

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Documentation Index

​Prerequisites

​Core Classes

BaseModel

SqlAlchemyRepository

​Step 1 — Define the ORM Model

​Step 2 — Define the Domain Entity

​Step 3 — Define the Repository

​Available Methods

​Field Resolvers and Serializers

​fields_resolvers — Model → Entity

​fields_serializers — Entity → Model

​Enabling Limit/Offset Pagination

​Session Factory

​Registering Models for Alembic

​Complete End-to-End Example

Build docs developers (and LLMs) love

Prerequisites

Core Classes

Step 1 — Define the ORM Model

Step 2 — Define the Domain Entity

Step 3 — Define the Repository

Available Methods

Field Resolvers and Serializers

`fields_resolvers` — Model → Entity

`fields_serializers` — Entity → Model

Enabling Limit/Offset Pagination

Session Factory

Registering Models for Alembic

Complete End-to-End Example