repository

Writing repositories (src/database/psql/repositories/)

Repositories are thin wrappers around CRUDRepository that add domain semantics (named arguments, loads, ordering, @on_integrity for unique constraints). The CRUD verbs are fixed — do not invent new method names.

Allowed method vocabulary — memorize this list

A repository may expose only these verbs (one per responsibility):

Method	Purpose	Return
create(**data)	insert one row	Result[M]
select(*loads, **filters)	fetch one row by identifier(s)	Result[M]
select_many(*loads, **filters, order_by, offset, limit)	paginated list	Result[tuple[int, Sequence[M]]]
update(uuid, /, **data)	update one row by id	Result[M]
delete(**filters)	delete one row by id	Result[M]
exists(**filters)	existence check	Result[bool]
count(**filters)	count matching rows	Result[int]
upsert(*conflict_cols, **data)	insert-or-update on conflict	Result[M]

Do not invent verbs like get, fetch, find_one, find_by_email, list_all, save, remove, get_or_create, paginate, search. If you need "find user by email", that's still select(login=...) with a new named parameter. If the current method doesn't support a filter you need, add a new keyword arg to the existing method — do not add a new method.

The only acceptable additions beyond this vocabulary are domain-specific bulk variants that mirror CRUD (insert_many on CRUDRepository already exists — use it via self._crud.insert_many(...)).

Anatomy of a repository

Every repository inherits BaseRepository[models.X] and accesses CRUD primitives through self._crud:

# src/database/psql/repositories/widget.py
from collections.abc import Sequence
from typing import Optional, Unpack

import uuid_utils.compat as uuid
from sqlalchemy import ColumnExpressionArgument

import src.database.psql.models as models
from src.database.psql.exceptions import InvalidParamsError
from src.database.psql.repositories import Result
from src.database.psql.repositories.base import BaseRepository
from src.database.psql.tools import (
    on_integrity,
    sqla_offset_query,
    sqla_select,
    unique_scalars,
)
from src.database.psql.types import OrderBy
from src.database.psql.types.widget import (
    CreateWidgetType,
    UpdateWidgetType,
    WidgetLoads,
)


class WidgetRepository(BaseRepository[models.Widget]):
    __slots__ = ()

    @on_integrity("name")
    async def create(self, **data: Unpack[CreateWidgetType]) -> Result[models.Widget]:
        return Result("create", await self._crud.insert(**data))

    async def select(
        self,
        *loads: WidgetLoads,
        widget_uuid: Optional[uuid.UUID] = None,
        name: Optional[str] = None,
    ) -> Result[models.Widget]:
        if not any([widget_uuid, name]):
            raise InvalidParamsError("at least one identifier must be provided")

        where_clauses: list[ColumnExpressionArgument[bool]] = []
        if widget_uuid:
            where_clauses.append(self.model.uuid == widget_uuid)
        if name:
            where_clauses.append(self.model.name == name)

        stmt = sqla_select(model=self.model, loads=loads).where(*where_clauses)
        return Result(
            "select", unique_scalars(await self._session.execute(stmt)).first()
        )

    @on_integrity("name")
    async def update(
        self,
        uuid: uuid.UUID,
        /,
        **data: Unpack[UpdateWidgetType],
    ) -> Result[models.Widget]:
        result = await self._crud.update(self.model.uuid == uuid, **data)
        return Result("update", result[0] if result else None)

    async def delete(
        self, widget_uuid: Optional[uuid.UUID] = None
    ) -> Result[models.Widget]:
        if not widget_uuid:
            raise InvalidParamsError("at least one identifier must be provided")

        result = await self._crud.delete(self.model.uuid == widget_uuid)
        return Result("delete", result[0] if result else None)

    async def select_many(
        self,
        *loads: WidgetLoads,
        name: Optional[str] = None,
        order_by: OrderBy = "desc",
        offset: int = 0,
        limit: Optional[int] = None,
    ) -> Result[tuple[int, Sequence[models.Widget]]]:
        where_clauses: list[ColumnExpressionArgument[bool]] = []

        if name:
            where_clauses.append(self.model.name.ilike(f"%{name}%"))

        total = await self._crud.count(*where_clauses)
        if total <= 0:
            return Result("select", (total, []))

        stmt = sqla_offset_query(
            self.model,
            loads=loads,
            offset=offset,
            limit=limit,
            order=("created_at", order_by),
            where=where_clauses,
        )
        results = unique_scalars(await self._session.execute(stmt)).all()
        return Result("select", (total, results))

    async def exists(self, name: str) -> Result[bool]:
        return Result("exists", await self._crud.exists(self.model.name == name))