repositron

A typed, generic repository base for SQLAlchemy 2.0. Full CRUD, zero per-table boilerplate.

Documentation · Filtering · Projection · Hooks · Sessions · API

---

Every SQLAlchemy project rewrites the same repository layer: a class per table wrapping select(...) / session.scalars(...), the same get / list / count, the same pagination math, the same "turn the ORM row into something light to return". It is mechanical, easy to get subtly wrong, and you write it again for the next table.

repositron writes that layer once, generically. Declare a model (and optionally a DTO and write payloads), inherit one class, and get a fully typed repository, every method checked against the types you declared.

from dataclasses import dataclass

from sqlalchemy.ext.asyncio import AsyncSession

from repositron import Repository, UNSET, UnsetType


@dataclass(frozen=True, slots=True)
class TaskDTO:                 # light, detached, serializes straight to JSON
    id: int
    title: str
    status: str
    assignee_id: int | None


@dataclass
class TaskCreate:
    workspace_id: int
    title: str


@dataclass
class TaskUpdate:
    title: str | UnsetType = UNSET            # absent = leave alone
    status: str | UnsetType = UNSET
    assignee_id: int | None | UnsetType = UNSET   # None = unassign (SET NULL)


class TaskRepository(
    Repository[Task, TaskDTO, TaskCreate, TaskUpdate, int, AsyncSession]
):
    pass                                       # the whole repository

That is the whole repository. get / first / list / list_paginated / count / exists / create / update / delete all come for free, typed against TaskDTO:

async def use_tasks(session: AsyncSession) -> None:
    repo = TaskRepository(session)

    task = await repo.get(1)  # TaskDTO | None
    tasks = await repo.list(workspace_id=42, status="open", order_by=Task.created_at.desc())
    page = await repo.list_paginated(0, 20, order_by=Task.created_at.desc())
    task_id = await repo.create(TaskCreate(workspace_id=42, title="Ship the docs"))
    updated = await repo.update(task_id, TaskUpdate(status="done"))

This is the default path for FastAPI and any asyncio application. The async guide covers session lifetime, relationship loading, hooks, and custom writes. For a conventional script or worker, bind the same base to Session and call get_sync, list_sync, and the other explicit sync methods.

Why repositron

It cuts the layer you keep rewriting. One generic base replaces the per-table CRUD class, and every method is typed off the generic parameters, so your editor knows await repo.list() returns list[TaskDTO] and repo.get(id) is checked against the key type you declared (int, str, uuid.UUID).

Two ways to filter, in one call. Equality is keyed by attribute name, anything else is a plain SQLAlchemy expression, and they combine. A None value means IS NULL; UNSET skips the filter, so optional query params pass straight through without branching.

await repo.list(workspace_id=42, extra_filters=[Task.archived_at.is_(None)], order_by=Task.id)
# WHERE workspace_id = 42 AND archived_at IS NULL ORDER BY id  (open, non-archived)

Updates that can actually write NULL. UNSET means "leave this column alone", None means "set it to NULL", the distinction the hand-written if x is not None pattern silently loses. TaskUpdate(assignee_id=None) unassigns a task; TaskUpdate(status="done") leaves the assignee untouched.

Projection that is real column selection. Index the repo with a narrow shape and it narrows the SELECT itself, it does not fetch the row and drop fields. The injected repository is untouched, the projection lasts only for the call.

@dataclass(frozen=True, slots=True)
class TaskCard:
    id: int
    title: str
    status: str

await repo[TaskCard].list(workspace_id=42, status="open")
# SELECT tasks.id, tasks.title, tasks.status FROM tasks
#   WHERE workspace_id = 42 AND status = 'open'
#   -> list[TaskCard]   (only those three columns ever leave the database)

Extend without overriding. Hooks layer a derived column, an enriched DTO, or an audit row onto the base, and @writes gives a custom write the same flush/commit/rollback the built-ins get, no self.session plumbing.

Your choice of DTO. A dataclass that serializes straight to JSON (so the same object is your repository return value and your FastAPI response_model), the model itself, or a Pydantic schema you already have.

Install

uv add "repositron[asyncio]"        # or: pip install "repositron[asyncio]"

Requires Python 3.13+ and sqlalchemy>=2.0. The asyncio extra installs SQLAlchemy's async support; Pydantic remains optional.

Documentation

Full guides and API reference at repositron.fa.dev.br.

License

MIT. See LICENSE.