s-ormkit

Generic БД/ORM-инфраструктура для экосистемы S-kits на SQLAlchemy 2.0: диалект-нейтральный движок из DB-URL, Repository + UnitOfWork + DIP-протоколы.

Версия: 0.0.1 (первый релиз)
Лицензия: MIT
Зависимости: SQLAlchemy>=2.0 (и ничего больше)

Кит НЕ знает ни о каком приложении. Прикладная логика получает БД-слой через ORM так, что не зависит от sqlite/диалекта: движок конфигурируется через DB-URL и заменяется на postgresql://.../mysql://... без изменения кода.

Что входит

Declarative-база и миксины (base.py)

• Base — общий declarative base для моделей-наследников
• IntPkMixin — целочисленный автоинкрементный первичный ключ
• TimestampMixin — created_at / updated_at через func.now() (диалект-нейтрально)

Движок из DB-URL (engine.py)

• make_engine — резолв URL (аргумент → env → default), для sqlite включает PRAGMA foreign_keys=ON; create_engine ленив (postgres-URL создаётся без коннекта)
• make_session_factory — sessionmaker(expire_on_commit=False)
• init_schema / drop_schema — создать/удалить таблицы
• dispose_engine — освободить пул соединений

Repository-паттерн (repository.py)

• BaseRepository[TModel, TDomain] — generic CRUD: add / get / get_or_none / list / update / remove / count
• Маппинг ORM <-> domain через переопределяемые хуки to_domain / to_orm
• Не коммитит сам (это делает UnitOfWork) — только flush для получения id

Транзакционная граница (unit_of_work.py)

• UnitOfWork — контекст-менеджер: rollback при исключении, close всегда, commit явный

DIP-контракты (protocols.py)

• RepositoryProtocol / UnitOfWorkProtocol — чтобы сервисы зависели от абстракций, а не от BaseRepository / SQLAlchemy

Быстрый старт

Кит generic — вы объявляете свои модели поверх Base и миксинов:

from dataclasses import dataclass

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

from ormkit import (
    Base, IntPkMixin, TimestampMixin,
    BaseRepository, UnitOfWork,
    make_engine, make_session_factory, init_schema,
)


# 1. Свои ORM-модели
class User(Base, IntPkMixin, TimestampMixin):
    __tablename__ = "users"
    name: Mapped[str] = mapped_column(String(100))


# 2. Свой доменный объект (dataclass / pydantic / dict — кит не знает тип)
@dataclass
class UserDTO:
    name: str
    id: int | None = None


# 3. Свой репозиторий — переопределяем ТОЛЬКО хуки маппинга
class UserRepository(BaseRepository[User, UserDTO]):
    def to_domain(self, obj: User) -> UserDTO:
        return UserDTO(id=obj.id, name=obj.name)

    def to_orm(self, domain: UserDTO) -> User:
        return User(name=domain.name)

Конфигурация движка через DB-URL

# sqlite по умолчанию (для локали / тестов), FK enforcement включён автоматически
engine = make_engine("sqlite:///app.db")

# та же строка кода на проде — просто другой URL, коду всё равно:
engine = make_engine(default="postgresql://user:pass@host/db")
# или через переменную окружения ORMKIT_DB_URL:
engine = make_engine()

init_schema(engine)
session_factory = make_session_factory(engine)

Работа через UnitOfWork

with UnitOfWork(session_factory) as uow:
    repo = UserRepository(uow.session, User)

    saved = repo.add(UserDTO(name="Алиса"))        # flush -> id проставлен
    found = repo.get(saved.id)                      # NotFoundError, если нет
    everyone = repo.list()                          # фильтр: repo.list(name="Алиса")
    repo.update(saved.id, name="Алиса Смит")
    total = repo.count()

    uow.commit()   # без commit транзакция не персистится
# исключение в блоке -> автоматический rollback; сессия всегда закрывается

DIP: сервис зависит от абстракции

from ormkit import RepositoryProtocol

def register_user(repo: RepositoryProtocol[UserDTO], name: str) -> UserDTO:
    return repo.add(UserDTO(name=name))
# в проде — UserRepository, в тестах — in-memory фейк, реализующий тот же протокол

Архитектура

ormkit/
├── __init__.py          # Public API + __version__
├── exceptions.py        # OrmKitError, NotFoundError
├── base.py              # Base, IntPkMixin, TimestampMixin
├── engine.py            # make_engine, make_session_factory, init/drop_schema, dispose
├── repository.py        # BaseRepository[TModel, TDomain]
├── unit_of_work.py      # UnitOfWork
└── protocols.py         # RepositoryProtocol, UnitOfWorkProtocol

Тестирование

uv sync --extra dev
uv run pytest -q          # зелёный, coverage >= 80%
uv run ruff check .       # чисто

Тесты объявляют собственные демо-модели (_User, _Post) прямо в conftest.py — это доказывает, что кит generic и не тянет никакого приложения.

Диалект-нейтральность

• Один код — любая СУБД. Меняется только DB-URL: sqlite:// ↔ postgresql:// ↔ mysql://. Прикладной код не трогается.
• sqlite FK enforcement. Для sqlite make_engine навешивает PRAGMA foreign_keys=ON (иначе sqlite молча игнорирует внешние ключи). Отключается флагом foreign_keys=False.
• Временные метки. TimestampMixin использует func.now(), работающий и в sqlite, и в postgres/mysql.
• Ленивое создание. make_engine("postgresql://...") не коннектится — движок создаётся без живого сервера.

Лицензия

MIT