Custom queries¶

The base class gives you CRUD and bulk writes. Real repositories grow past both: a free-text search, an upsert, a query that joins three tables to answer one question. repositron's job is to remove the boilerplate, not to box you in, so everything on your repository is an ordinary class with self.session and self.model_class to build on.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from repositron import Repository, UNSET, UnsetType


class Base(DeclarativeBase): ...


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    workspace_id: Mapped[int]
    title: Mapped[str]
    description: Mapped[str | None] = mapped_column(default=None)
    status: Mapped[str] = mapped_column(default="open")
    assignee_id: Mapped[int | None] = mapped_column(default=None)
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    archived_at: Mapped[datetime | None] = mapped_column(default=None)


class Member(Base):
    __tablename__ = "members"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]


class Subtask(Base):
    __tablename__ = "subtasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    task_id: Mapped[int] = mapped_column(ForeignKey("tasks.id"))
    title: Mapped[str]


@dataclass
class TaskDTO:
    id: int
    title: str
    status: str
    assignee_id: int | None


@dataclass
class TaskCreate:
    workspace_id: int
    title: str
    description: str | None | UnsetType = UNSET
    assignee_id: int | None | UnsetType = UNSET


@dataclass
class TaskUpdate:
    title: str | UnsetType = UNSET
    description: str | None | UnsetType = UNSET
    status: str | UnsetType = UNSET
    assignee_id: int | None | UnsetType = UNSET


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

Domain queries¶

A method that does not fit get / list is just a method. You have the session, the model, and the full SQLAlchemy API:

The examples use AsyncSession. Await SQLAlchemy I/O and the inherited repository methods; the synchronous counterpart uses the same logic with its explicit _sync calls.

from sqlalchemy import func, select


class TaskRepository(Repository[Task, TaskDTO, TaskCreate, TaskUpdate, int, AsyncSession]):
    async def open_in_workspace(self, workspace_id: int) -> list[TaskDTO]:
        return await self.list(status="open", workspace_id=workspace_id)

    async def status_counts(self, workspace_id: int) -> dict[str, int]:
        rows = (await self.session.execute(
            select(Task.status, func.count())
            .where(Task.workspace_id == workspace_id)
            .group_by(Task.status)
        )).all()
        return dict(rows)

Note the first method reuses await self.list instead of reaching for the session. Build on the inherited methods where they fit; drop to raw SQLAlchemy only where they do not. A plain bulk update or delete is one of the cases that does fit, update_where / delete_where cover it without a custom method.

Filter builders¶

When the same WHERE fragment shows up in several calls, a free-text search being the classic case, give it a name. A method that returns a SQLAlchemy expression plugs straight into extra_filters:

from sqlalchemy import ColumnElement, or_


class TaskRepository(Repository[Task, TaskDTO, TaskCreate, TaskUpdate, int, AsyncSession]):
    def search(self, q: str) -> ColumnElement[bool]:
        pattern = f"%{q}%"
        return or_(
            Task.title.ilike(pattern),
            Task.description.ilike(pattern),
        )


await repo.list(extra_filters=[repo.search("deploy")], status="open")

Callers express intent ("search for deploy") and the column logic lives in one place.

Custom hydration¶

The automatic model-to-DTO conversion handles the common cases: a dataclass built by field name, a Pydantic model through model_validate, or the model returned as-is.

For the rest, the question is add or replace. In an async application, a hook is the right way to add a value that needs database I/O:

• To add a derived field to the built DTO, use a hydrate hook. It hands you the finished DTO to enrich, so you write one field, not all of them.
• To replace the build, when the automatic path cannot produce the DTO at all, use a build hook or override _hydrate with a construction that does not issue I/O.

from dataclasses import dataclass

from sqlalchemy import select

from repositron import ReadOnlyRepository, on


@dataclass
class TaskDetail:
    id: int
    title: str
    status: str
    assignee_name: str | None   # rolled up from Member, not a column on Task


class TaskRepository(ReadOnlyRepository[Task, TaskDetail, int, AsyncSession]):
    @on("hydrate", mode="after")
    async def add_assignee_name(self, model: Task, dto: TaskDetail) -> TaskDetail:
        dto.assignee_name = await self.session.scalar(
            select(Member.name).where(Member.id == model.assignee_id)
        )
        return dto

The hook runs for every read, so get, first, and list all return a fully-formed TaskDetail. Do not use this exact pattern for a long list: it would make one extra query per row. Write a custom query that preloads or aggregates the related data instead. (Column projection via repo[Shape] builds the narrow shape positionally and does not run hydrate hooks.)

Overriding _hydrate and tagging a method with @on("hydrate", mode="build") are the same mechanism, the override is just the build hook spelled as a method. Use either when construction needs no database work; use an async after-hook when it does.

Transactions on custom writes¶

A custom write is responsible for the same flush / commit / rollback dance the built-in create / update / delete handle for you. @writes gives a custom method that dance, so its body is only the session work. Reach for it when the write is past what bulk writes cover, an upsert here:

from sqlalchemy.dialects.postgresql import insert

from repositron import Repository, writes


class TaskRepository(Repository[Task, TaskDTO, TaskCreate, TaskUpdate, int, AsyncSession]):
    @writes
    async def upsert_by_title(self, workspace_id: int, title: str, status: str) -> None:
        stmt = insert(Task).values(workspace_id=workspace_id, title=title, status=status)
        await self.session.execute(
            stmt.on_conflict_do_update(index_elements=["title"], set_={"status": status})
        )   # flushed for you; rolled back on error

The decorated method flushes after the body, commits if the repository is autocommit=True, and rolls back on error, exactly like the built-in writes (see committing). To let a caller commit a single write, declare a commit parameter and @writes honors it:

    @writes
    async def upsert_by_title(
        self, workspace_id: int, title: str, status: str, *, commit: bool | None = None
    ) -> None:
        stmt = insert(Task).values(workspace_id=workspace_id, title=title, status=status)
        await self.session.execute(
            stmt.on_conflict_do_update(index_elements=["title"], set_={"status": status})
        )


await repo.upsert_by_title(1, "Ship docs", "open", commit=True)

When the method needs the primary key mid-way, to attach child rows or return it, flush yourself at that point. @writes still owns the final flush and the commit/rollback:

    @writes
    async def create_with_subtasks(self, payload: TaskCreate, subtasks: list[str]) -> int:
        task = Task(workspace_id=payload.workspace_id, title=payload.title)
        self.session.add(task)
        await self.session.flush()  # need task.id for the subtasks below
        for title in subtasks:
            self.session.add(Subtask(task_id=task.id, title=title))
        return task.id

To return a DTO from a custom write instead of an id, hydrate the saved row with await self._hydrate_one_async, so a custom build or hydrate after-hooks apply:

    @writes
    async def create_with_subtasks(self, payload: TaskCreate, subtasks: list[str]) -> TaskDTO:
        task = Task(workspace_id=payload.workspace_id, title=payload.title)
        self.session.add(task)
        await self.session.flush()
        for title in subtasks:
            self.session.add(Subtask(task_id=task.id, title=title))
        return await self._hydrate_one_async(task)

Without @writes, a custom write should still flush, never commit, the same as the base class, so it composes inside the caller's transaction. See the design principles.