Primary keys¶

Not every table is keyed by an auto-incrementing id. repositron handles the common variations, a different column name, a str or uuid key, without ceremony, and types the id arguments along the way.

The default: an integer id¶

Out of the box, a repository assumes the key column is named id and holds an int. Every id-based method, get, update, delete, exists, is typed to take an int, and create returns one:

from __future__ import annotations

from dataclasses import dataclass

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

from repositron import ReadOnlyRepository, 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]
    status: Mapped[str] = mapped_column(default="open")


@dataclass
class TaskDTO:
    id: int
    title: str
    status: str


@dataclass
class TaskCreate:
    workspace_id: int
    title: str


@dataclass
class TaskUpdate:
    title: str | UnsetType = UNSET
    status: str | UnsetType = UNSET


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

repo = TaskRepository(session)

await repo.get(1)         # TaskDTO | None
await repo.exists(1)      # bool
await repo.get("1")       # type error: expected int, got str

That last line is the point: because the key type is part of the repository's type, a checker (pyright, mypy, ty) catches a stringly-typed id before it runs. For the 90% of tables keyed by an int, you get this for free, no extra declaration.

A key that is not an int¶

When the key is a str, a uuid, or anything else, declare its type in the PK slot. On Repository, it is the fifth parameter, immediately before the optional SessionT; on ReadOnlyRepository, it is the third. SessionT defaults to synchronous Session, so the common case still does not mention it.

On a read-only repository it is the third parameter. A Page cache, keyed by the hash of a fetched URL, is a natural str key:

class Page(Base):
    __tablename__ = "pages"

    url_hash: Mapped[str] = mapped_column(primary_key=True)
    html: Mapped[str]


@dataclass
class PageDTO:
    url_hash: str
    html: str


class PageRepository(ReadOnlyRepository[Page, PageDTO, str, AsyncSession]):
    pk_column = "url_hash"


await repo.get("a1b2c3")  # PageDTO | None
await repo.get(123)       # type error: expected str, got int

A uuid key on a full-CRUD repository is no different, name uuid.UUID as the type. An API token keyed by a generated token_id is the typical case:

import uuid


class ApiToken(Base):
    __tablename__ = "api_tokens"

    token_id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    workspace_id: Mapped[int]
    label: Mapped[str]


@dataclass
class ApiTokenDTO:
    token_id: uuid.UUID
    label: str


@dataclass
class ApiTokenCreate:
    workspace_id: int
    label: str


@dataclass
class ApiTokenUpdate:
    label: str | UnsetType = UNSET


class ApiTokenRepository(
    Repository[ApiToken, ApiTokenDTO, ApiTokenCreate, ApiTokenUpdate, uuid.UUID, AsyncSession]
):
    pk_column = "token_id"


tid = await repo.create(ApiTokenCreate(workspace_id=1, label="ci"))  # uuid.UUID
await repo.get(tid)                                                  # ApiTokenDTO | None

The slots are positional¶

Type parameters cannot be passed by name, so to reach PKT you fill the ones before it. SessionT is the final slot. If a CRUD repository has real Create/Update payloads, declare them; that is the honest signature anyway:

Repository[ApiToken, ApiTokenDTO, ApiTokenCreate, ApiTokenUpdate, uuid.UUID]

The examples in this guide bind that final slot to AsyncSession, which makes the inherited calls awaitable:

from sqlalchemy.ext.asyncio import AsyncSession


class AsyncApiTokenRepository(
    Repository[
        ApiToken,
        ApiTokenDTO,
        ApiTokenCreate,
        ApiTokenUpdate,
        uuid.UUID,
        AsyncSession,
    ]
):
    pk_column = "token_id"

For a synchronous repository, bind the same final slot to Session and use the explicit _sync methods. Padding the payload slots with object only makes sense when you genuinely don't use create/update. An append-only AuditEntry is written once and never updated, so it has no update payload and the slot is padded:

class AuditEntry(Base):
    __tablename__ = "audit_entries"

    trace_id: Mapped[str] = mapped_column(primary_key=True)
    action: Mapped[str]


@dataclass
class AuditEntryDTO:
    trace_id: str
    action: str


@dataclass
class AuditEntryCreate:
    trace_id: str
    action: str


# entries are created and read, never updated, so Update stays untyped
class AuditEntryRepository(
    Repository[AuditEntry, AuditEntryDTO, AuditEntryCreate, object, str]
):
    pk_column = "trace_id"

If you want typed creates and updates, declare them properly rather than reaching for object.

When the key is not called id¶

Independent of its type, point pk_column at the right column. Reusing the Page cache from above, two forms work:

class PageRepository(ReadOnlyRepository[Page, PageDTO, str]):
    pk_column = "url_hash"      # by name


class PageRepository(ReadOnlyRepository[Page, PageDTO, str]):
    pk_column = Page.url_hash   # by column reference

The string is terse; the column reference reads naturally next to the rest of your SQLAlchemy code. Either way the column is resolved by name through the model on first use, so a column from the wrong model (or a bad name) raises AttributeError at the first query rather than cross-joining. Pick whichever you prefer, they resolve to the same column. Both feed every id-based method:

await repo.get("a1b2c3")    # WHERE url_hash = 'a1b2c3'
await repo.exists("a1b2c3") # same column

This is the case for tables keyed by a natural identifier, a hash, a slug, an external system's id, rather than a surrogate id.

Filtering on the key like any other column¶

pk_column only affects the id-based methods. The key column is still an ordinary attribute, so you can filter on it through the normal channels when you want a query rather than a single fetch:

await repo.get("a1b2c3")    # one PageDTO | None, keyed lookup

hashes = ["a1b2c3", "d4e5f6", "789abc"]
await repo.list(extra_filters=[Page.url_hash.in_(hashes)])  # list[PageDTO]

That returns a list, where get returns one. Reach for whichever matches what you actually need.

What about composite keys?¶

repositron's id-based methods assume a single key column. A table with a composite primary key still works as a repository; you simply do not use get / update(id, ...) / delete(id) against it. A Membership association keyed by (workspace_id, member_id) is the classic shape. Filter on the key columns and add your own methods for the lookups:

from sqlalchemy import select


class Membership(Base):
    __tablename__ = "memberships"

    workspace_id: Mapped[int] = mapped_column(primary_key=True)
    member_id: Mapped[int] = mapped_column(primary_key=True)
    role: Mapped[str] = mapped_column(default="member")


@dataclass
class MembershipDTO:
    workspace_id: int
    member_id: int
    role: str


class MembershipRepository(Repository[Membership, MembershipDTO]):

    def get_by_key_sync(self, workspace_id: int, member_id: int) -> MembershipDTO | None:
        model = self.session.scalars(
            select(Membership)
            .where(Membership.workspace_id == workspace_id)
            .where(Membership.member_id == member_id)
        ).first()
        return self._hydrate_one_sync(model) if model is not None else None

Do not name this method get, which is the awaited single-key API. The async version follows the same query shape, awaits self.session.execute(...), and finishes with await self._hydrate_one_async(model).

The read and filter machinery, list, first, count, extra_filters, and projection is indifferent to how many columns make up the key.