Skip to main content

Repository with mixins, templates, and utilities for work with SQLAlchemy

Project description

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

Tool Purpose
Poetry Dependency management
Ruff Linting & formatting
mypy Static type checking
pytest Testing framework

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

alchemist_cerber-0.1.0.tar.gz (8.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

alchemist_cerber-0.1.0-py3-none-any.whl (10.1 kB view details)

Uploaded Python 3

File details

Details for the file alchemist_cerber-0.1.0.tar.gz.

File metadata

  • Download URL: alchemist_cerber-0.1.0.tar.gz
  • Upload date:
  • Size: 8.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.2 CPython/3.12.12 Linux/6.11.0-1018-azure

File hashes

Hashes for alchemist_cerber-0.1.0.tar.gz
Algorithm Hash digest
SHA256 2051a678d055f64d6d4ec1710d15d50960112ec7d28edbecaf7f13759cb3750f
MD5 e44b82b35ac9fcce7037c5733a6ba6e3
BLAKE2b-256 5f66846d5a49738977fe329d11ceafd6bd57e3a1077e6a04fa2e3a153a79d300

See more details on using hashes here.

File details

Details for the file alchemist_cerber-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: alchemist_cerber-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 10.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.2 CPython/3.12.12 Linux/6.11.0-1018-azure

File hashes

Hashes for alchemist_cerber-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 3e129ea1d85732c1000c012b06e34fd331f01ed45fd7c013077b9992c67074e3
MD5 8c8e3dd7f45c5b9cdce2a0fa6adde0eb
BLAKE2b-256 5e387ee951b6a2eb478ff7b1a13b472469a1637b9ff9d5cc331a092a1ca8cba2

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page