repositron¶

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

Get started Sessions and execution All guides

For agents and documentation tools

Start with the machine-readable documentation index. It maps implementation questions to the smallest relevant guide and states the awaited calling convention.

Every SQLAlchemy project ends up with the same folder: one repository class per table, each wrapping select(...) / session.scalars(...) in the same get, the same list, the same pagination math. repositron writes that layer once, generically, and types it against your model and your return shape.

Hand-written, per table Declared once

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession


class TaskRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get(self, id: int) -> TaskDTO | None:
        row = (await self.session.scalars(select(Task).where(Task.id == id))).first()
        if row is None:
            return None
        return TaskDTO(id=row.id, title=row.title, status=row.status, assignee_id=row.assignee_id)

    async def list(self, *, status: str | None = None) -> list[TaskDTO]:
        stmt = select(Task)
        if status is not None:
            stmt = stmt.where(Task.status == status)
        rows = (await self.session.scalars(stmt)).all()
        return [TaskDTO(id=r.id, title=r.title, status=r.status, assignee_id=r.assignee_id) for r in rows]

    async def update(self, id: int, *, assignee_id: int | None = None) -> bool:
        task = (await self.session.scalars(select(Task).where(Task.id == id))).first()
        if task is None:
            return False
        if assignee_id is not None:   # and how do you unassign on purpose?
            task.assignee_id = assignee_id
        await self.session.flush()
        return True

    # ...count, delete, first, pagination, then again for the next ten tables.

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 leaves it; None sets NULL
    status: str | UnsetType = UNSET
    assignee_id: int | None | UnsetType = UNSET


class TaskRepository(
    Repository[Task, TaskDTO, TaskCreate, TaskUpdate, int, AsyncSession]
):
    ...

Every method from the other tab now exists, typed against TaskDTO, with no further code.

What you get¶

Typed end to end

await repo.list() is list[TaskDTO], and your editor knows it. No casts, no Any. The return value is the same object your API serializes.
Two ways to filter, one call

Equality by keyword and arbitrary SQLAlchemy expressions, combined. You never pick between readable and powerful.

Filtering
Updates that write NULL on purpose

UNSET leaves a column alone; None sets it to NULL. The is not None pattern cannot tell those apart. repositron can.

Updating rows
Load only what you need

await repo[Card].list() selects just that shape's columns, for one call, without touching the injected repository.

Projection
Pagination that refuses to lie

list_paginated requires order_by and raises if you forget, turning a production heisenbug into an error at the call site.

Pagination
One dependency

Just sqlalchemy>=2.0. Dataclass DTOs add nothing else; Pydantic is detected only if your DTO is one.

Install¶

uvpip

uv add repositron

pip install repositron

Python 3.13+ and sqlalchemy>=2.0.

Get started