# Repository

> Use when creating or extending a repository under src/database/psql/repositories/. Enforces the project's fixed repository vocabulary (create, select, select_many, update, delete, exists, count, upsert) — no invented methods. Shows exactly how to wrap CRUDRepository, use @on_integrity, build loads-aware selects, and return Result[T].

- **Type:** Skill
- **Install:** `agentstack add skill-ocbunknown-fastapi-claude-template-repository`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [ocbunknown](https://agentstack.voostack.com/s/ocbunknown)
- **Installs:** 0
- **Category:** [Databases](https://agentstack.voostack.com/c/databases)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [ocbunknown](https://github.com/ocbunknown)
- **Source:** https://github.com/ocbunknown/fastapi-claude-template/tree/master/.claude/skills/repository

## Install

```sh
agentstack add skill-ocbunknown-fastapi-claude-template-repository
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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`:

```python
# 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  Result[bool]:
        return Result("exists", await self._crud.exists(self.model.name == name))
```

## Fixed rules

1. **Always return `Result[T]`** from every public method. The first argument to `Result(...)` is the exception key (`"create" | "select" | "select_many" | "update" | "delete" | "exists" | "count" | "upsert"`) — it determines which domain exception `result.result()` raises on `None`.
2. **Always use `sqla_select(model=..., loads=loads)`** for single-row reads with eager loading and **`sqla_offset_query(model, loads=loads, offset=..., limit=..., order=(col, dir), where=...)`** for paginated reads. Both come from `src.database.psql.tools` (re-exported from `sqla-autoloads`). Never write raw `select(...).options(selectinload(...))` in a repository, and never apply `.limit()/.offset()` directly to a `sqla_select` query — the library helper handles pagination correctly via a CTE on the primary key so eager-loading joins operate only on the page slice.
3. **Always materialise results via `unique_scalars(...)`** (also re-exported from `src.database.psql.tools`). It returns a `ScalarResult` so you can chain `.first()` for single rows or `.all()` for collections. This deduplicates rows produced by outer-join eager loading — never call `.unique().scalars()` by hand.
4. **Always decorate `create` / `update` / `upsert` with `@on_integrity("unique_col_1", "unique_col_2")`** if the model has unique constraints. This converts SQLAlchemy `IntegrityError` into `ConflictError(" already in use")`.
5. **`select_many` must honour `limit: Optional[int] = None`** — no upper cap, no default limit. The cap is enforced at the contract layer (`presentation/http/v1/contracts/pagination.py` — 200 max). Internal callers (use cases, tasks) can pass `limit=None` for "all rows". `select_many` also returns `(total, rows)` — total is always computed via `self._crud.count(*where_clauses)` **before** the rows query, and if `total  limit
items = rows[:limit]
```

## Types & loads (required companion files)

For every new repository, add (or extend) `src/database/psql/types/.py`:

```python
from typing import Literal, TypedDict
from uuid_utils.compat import UUID

WidgetLoads = Literal["owner", "tags"]  # relationships the caller may eager-load

class CreateWidgetType(TypedDict):
    name: str
    owner_uuid: UUID

class UpdateWidgetType(TypedDict, total=False):
    name: str | None
    owner_uuid: UUID | None
```

`WidgetLoads` lists the relationship names (str literals) that `sqla_select` knows how to eager-load. Only include relationships that are actually defined on the model.

## Wire into `DBGateway`

After writing the repository and types, add a property on `src/database/psql/__init__.py::DBGateway`:

```python
@property
def widget(self) -> WidgetRepository:
    return self._from_cache("widget", WidgetRepository, model=models.Widget)
```

The `_from_cache` helper memoizes the instance per gateway (= per request scope), so multiple use cases in the same request share one repository/one session.

## Anti-patterns (hard no)

- ❌ Inventing new verb names (`find_by_name`, `list_active`, `bulk_create`, `save`, `remove`, `get_or_create`, `paginate`, `search`) — use existing CRUD verbs with new kwargs.
- ❌ Calling `self._session.execute(text(...))` or writing raw SQL strings inside a repository — use Query Objects under `database/psql/queries/` for complex SQL.
- ❌ Committing inside a repository method (`self._session.commit()`) — commits are owned by the `DBGateway` context manager.
- ❌ Returning a bare ORM model (`-> models.Widget`) instead of `Result[models.Widget]`.
- ❌ Hardcoding a `limit` default in `select_many` (e.g. `limit: int = 10`) — must be `Optional[int] = None`. Default pagination lives in the contract layer.
- ❌ Calling `UserResult.model_validate(...)` inside a repository. Result types live in `application/` — repositories never construct them.
- ❌ Adding `flush` / `refresh` / `expire` calls — the transaction manager handles that.
- ❌ Importing from `src.application.*` (use case layer) or `src.presentation.*` / `src.infrastructure.*`. The only legal cross-layer import is `src.application.common.exceptions` (and only for exceptions).

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [ocbunknown](https://github.com/ocbunknown)
- **Source:** [ocbunknown/fastapi-claude-template](https://github.com/ocbunknown/fastapi-claude-template)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-ocbunknown-fastapi-claude-template-repository
- Seller: https://agentstack.voostack.com/s/ocbunknown
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.