s-ormkit

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

Версия: 0.2.0
Лицензия: MIT
Зависимости: SQLAlchemy>=2.0 (alembic — опционально: s-ormkit[migrations])

Кит НЕ знает ни о каком приложении. Прикладная логика получает БД-слой через 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
• AsyncRepositoryProtocol / AsyncUnitOfWorkProtocol — то же для async-слоя

Async-слой (async_engine.py, async_repository.py, async_unit_of_work.py)

• make_async_engine / make_async_session_factory — асинхронный движок и async_sessionmaker(expire_on_commit=False); для sqlite так же включается PRAGMA foreign_keys=ON, create_async_engine так же ленив
• init_schema_async / drop_schema_async / dispose_engine_async
• AsyncUnitOfWork / AsyncBaseRepository — асинхронные близнецы с ТЕМ ЖЕ API

Резолв DB-URL (url.py)

• resolve_db_url — аргумент → переменная окружения → default (общий для sync, async и миграций)
• async_driver_url / sync_driver_url — автоподстановка драйвера (postgresql:// → postgresql+asyncpg:// или postgresql+psycopg://)

Защита от N+1 (loading.py)

• LoadSpec(selectin=, joined=) + apply_load_spec — декларация связей, которые грузятся вместе с сущностью (в т.ч. вложенные пути "posts.comments")
• strict_loading / enable_strict_loading / make_strict_async_session_factory — строгий режим raiseload("*"): незадекларированная ленивая подгрузка падает

Миграции (migrations.py, extra [migrations])

• scaffold_alembic и CLI python -m ormkit migrations init — генерация рабочего alembic-окружения на async-движке кита

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

Кит 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 фейк, реализующий тот же протокол

Асинхронное приложение

Тот же код, но на async/await — форма вызова другая, структура та же:

from ormkit import (
    AsyncBaseRepository, AsyncUnitOfWork,
    init_schema_async, make_async_engine, make_async_session_factory,
)


class UserRepository(AsyncBaseRepository[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)


# sqlite:// → sqlite+aiosqlite://, postgresql:// → postgresql+asyncpg:// (автоматически)
engine = make_async_engine("postgresql://user:pass@host/db")
await init_schema_async(engine)          # в проде схему катает alembic, см. ниже
session_factory = make_async_session_factory(engine)

async with AsyncUnitOfWork(session_factory) as uow:
    repo = UserRepository(uow.session, User)
    saved = await repo.add(UserDTO(name="Алиса"))
    await uow.commit()

Миграции (alembic)

create_all годится до первого изменения схемы на живых данных — дальше нужны миграции:

pip install "s-ormkit[migrations]"
python -m ormkit migrations init . --metadata myapp.models:Base
# дальше — штатный alembic, URL берётся из ORMKIT_DB_URL (как и у движка приложения):
alembic revision --autogenerate -m "init"
alembic upgrade head

Сгенерированный env.py работает на async-движке кита и с render_as_batch=True (без него sqlite не переживает ALTER). Ссылка на metadata (pkg.module:attr) — единственное, что нужно указать.

Защита от N+1

Репозиторий ДЕКЛАРИРУЕТ, что грузится вместе с сущностью, а строгий режим не даёт незаметно уехать в ленивую подгрузку:

from ormkit import LoadSpec, enable_strict_loading, make_strict_async_session_factory


class UserRepository(AsyncBaseRepository[User, UserDTO]):
    load_spec = LoadSpec(selectin=("posts.comments",), joined=("profile",))
    ...


# в conftest тестов / в CI: любая НЕобъявленная ленивая подгрузка → InvalidRequestError
session_factory = make_strict_async_session_factory(engine)
# либо точечно на уже открытой сессии (возвращает выключатель):
disable = enable_strict_loading(session)

selectin — для коллекций (отдельный IN-запрос), joined — для «многие-к-одному» (один JOIN). N+1 падает на этапе тестов, а не в проде.

Архитектура

ormkit/
├── __init__.py          # Public API + __version__
├── __main__.py          # CLI: python -m ormkit migrations init
├── exceptions.py        # OrmKitError, NotFoundError
├── base.py              # Base, IntPkMixin, TimestampMixin
├── url.py               # resolve_db_url, async_driver_url, sync_driver_url
├── engine.py            # make_engine, make_session_factory, init/drop_schema, dispose
├── async_engine.py      # make_async_engine, make_async_session_factory, *_async
├── repository.py        # BaseRepository[TModel, TDomain]
├── async_repository.py  # AsyncBaseRepository[TModel, TDomain]
├── unit_of_work.py      # UnitOfWork
├── async_unit_of_work.py# AsyncUnitOfWork
├── loading.py           # LoadSpec, apply_load_spec, строгий режим (raiseload)
├── migrations.py        # scaffold_alembic, migration_url, require_alembic
├── pagination.py        # Page, page_to_offset, paginate_stmt
├── search.py            # trigram_filter
├── rls.py               # TenantContext, apply_tenant_guc, rls_bypass
└── protocols.py         # RepositoryProtocol, UnitOfWorkProtocol (+ async)

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

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://...") не коннектится — движок создаётся без живого сервера. То же верно для make_async_engine.
• Драйвер подставляется сам. postgresql:// становится postgresql+asyncpg:// в async-контексте и postgresql+psycopg:// в синхронном; явно заданный драйвер не трогается.

Лицензия

MIT