Alchemist CeRBeR library

This repository layer implements a flexible, reusable, and type-safe ORM interaction system using SQLAlchemy, Python type hints*, and mixin-based composition. It supports both synchronous and asynchronous operations and provides a clean abstraction over database queries.

📦 Key Features

• Mixin-based architecture for reusable logic
• Synchronous and asynchronous implementations
• Type-safe ORM interactions using Pydantic and SQLAlchemy ORM
• Sorting, filtering, and joining capabilities
• Error handling with custom exceptions
• Unit of Work pattern for transaction management
• Support for filters using dynamic operators (eq, lt, in, between, etc.)

📚 Usage Example

1. Define ORM Models

import uuid
from datetime import UTC, datetime

from sqlalchemy import DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(primary_key=True, default=lambda: str(uuid.uuid4()))
    name: Mapped[str] = mapped_column(nullable=False)
    age: Mapped[int] = mapped_column(nullable=True)
    email: Mapped[str] = mapped_column(nullable=False, unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.now(UTC), nullable=False)

    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[str] = mapped_column(primary_key=True, default=lambda: str(uuid.uuid4()))
    title: Mapped[str] = mapped_column(nullable=False)
    content: Mapped[str] = mapped_column(nullable=False)
    author_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False)

    author: Mapped["User"] = relationship(back_populates="posts")

2. Build a Repository

from alchemist_cerber.mixins.readable import AsyncRepositoryGetMultiMixin, AsyncRepositoryGetOneMixin
from alchemist_cerber.utils.base import OrmModelT
from sqlalchemy.ext.asyncio import AsyncSession


class UserRepository(
    AsyncRepositoryGetMultiMixin[User],
    AsyncRepositoryGetOneMixin[User],
):
    _orm_model = User

    def __init__(self, session: AsyncSession):
        super().__init__(session)

3. Use the Repository

from alchemist_cerber.schemas import ColumnFilter, SortingItem
from alchemist_cerber.constants import AlchemyOperator

filters = [
    ColumnFilter(column=User.age, value=18, operator=AlchemyOperator.GT),
    ColumnFilter(column=User.name, value="Alice", operator=AlchemyOperator.EQ),
]

sorting = [SortingItem(field_name="age", is_asc=False)]

users = await user_repo.get_multi(
    limit=10,
    offset=0,
    orm_filters=tuple(filters),
    sort_params=tuple(sorting),
)

4. Configure Join Related Models

Override _use_join() in your repository to eager-load related data:

from sqlalchemy.orm import joinedload
from sqlalchemy import Select


class PostRepository(AsyncRepositoryGetMultiMixin[Post]):
    _orm_model = Post

    @staticmethod
    def _use_join(current_statement: Select[tuple[Post]]) -> Select[tuple[Post]]:
        return current_statement.options(joinedload(Post.author))

5. Use Unit of Work for Transactions

from repository.units_of_work import AsyncUnitOfWorkManager

async with AsyncUnitOfWorkManager(session_factory) as uow:
    user_repo = UserRepository(uow.session)
    post_repo = PostRepository(uow.session)

    user = await user_repo.get_one_unique(...)
    await post_repo.save(...)
    await uow.commit()

🛠️ Development Setup

This project uses Poetry for dependency management, Ruff for linting and formatting, mypy for type checking, and pytest for testing.

1. Install Dependencies

To install all required dependencies and set up the environment:

make init

This will:

• Install all dependencies via poetry install
• Copy .env.example to .env

2. Formatting Code

To format and sort imports:

make format

This will:

• Sort pyproject.toml keys
• Fix import order with ruff
• Auto-format code using ruff format

3. Linting & Type Checking

To lint and type-check the code:

make lint

This will:

• Run ruff for linting (flake8-compatible, fast)
• Run mypy for static type checking

4. Running Tests

To run tests:

make test

This will:

• Run all tests using pytest

🧰 Tools Used