repositron

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

Documentation · Filtering · Projection · Hooks · 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 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]):
    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:

repo = TaskRepository(session)  # a synchronous SQLAlchemy Session

repo.get_sync(1)                                         # -> TaskDTO | None
repo.list_sync(workspace_id=42, status="open", order_by=Task.created_at.desc())
repo.list_paginated_sync(0, 20, order_by=Task.created_at.desc())
repo.create_sync(TaskCreate(workspace_id=42, title="Ship the docs"))  # -> int (new id)
repo.update_sync(1, TaskUpdate(status="done"))          # only status; title untouched

For FastAPI or another async application, bind the same base to AsyncSession and await the unsuffixed methods: await repo.get(id), await repo.list(...), and await repo.create(payload). The full async and sync guide shows both session types, hooks, custom writes, and the migration mapping.

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.

repo.list_sync(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

repo[TaskCard].list_sync(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        # or: pip install repositron

Requires Python 3.13+ and sqlalchemy>=2.0, the only dependency.

Documentation

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

License

MIT. See LICENSE.