Metadata-Version: 2.4
Name: fast-repository
Version: 0.2.1
Summary: Interface-first repository pattern for FastAPI + SQLAlchemy. Declare the interface, get the implementation for free.
Project-URL: Homepage, https://github.com/2u2s/fast-repository
Author-email: 2u2s <2u2s.hello@google.com>
License-Expression: MIT
License-File: LICENSE
Keywords: clean-architecture,crud,fastapi,repository,repository-pattern,sqlalchemy
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: fastapi-pagination>=0.12
Requires-Dist: sqlalchemy[asyncio]>=2.0
Description-Content-Type: text/markdown

# fast-repository

**English** | [한국어](https://github.com/2u2s/fast-repository/blob/main/README.ko.md)

[![PyPI](https://img.shields.io/pypi/v/fast-repository)](https://pypi.org/project/fast-repository/)
[![Python](https://img.shields.io/pypi/pyversions/fast-repository)](https://pypi.org/project/fast-repository/)
[![License](https://img.shields.io/pypi/l/fast-repository)](https://github.com/2u2s/fast-repository/blob/main/LICENSE)
[![Downloads](https://static.pepy.tech/badge/fast-repository/month)](https://pepy.tech/project/fast-repository)

Interface-first repository pattern for FastAPI + SQLAlchemy.

Declare the repository interface, get the implementation for free.

## Why

FastAPI has no built-in repository concept. Its tutorials reach the database directly
inside path operation functions, or through a loose `crud.py` module of free functions.
That is fine for a demo, but as an app grows it scatters query logic across routes and
couples your business logic to SQLAlchemy.

The repository pattern puts one object in charge of persistence for an entity, and that
buys you:

- **A domain layer that depends on an interface, not on SQLAlchemy.** Business
  logic talks to `UserRepositoryInterface` — it never imports a session or writes
  a `select()`.
- **Tests without a database.** Swap the real repository for an in-memory fake at
  the interface boundary to unit-test business rules.
- **Query logic in one place.** Filtering, eager-loading, and soft-delete rules
  stop being copy-pasted across endpoints.
- **Swappable implementations.** Change the ORM, add caching, or split reads and
  writes without touching callers.

The catch: writing that repository for every entity — the various read, save, and
delete methods, and pagination on top — is the same boilerplate each time.

**Before** — hand-written, and repeated for every entity:

```python
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase): ...


class User(Base):
    __tablename__ = "users"

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


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

    async def find(self, id: int) -> User | None:
        return await self.session.scalar(select(User).where(User.id == id))

    async def find_all(
        self,
        name: str | None = None,
        age: int | None = None,
        status: str | None = None,
    ) -> list[User]:
        stmt = select(User)
        if name is not None:
            stmt = stmt.where(User.name == name)
        if age is not None:
            stmt = stmt.where(User.age == age)
        if status is not None:
            stmt = stmt.where(User.status == status)
        return list((await self.session.scalars(stmt)).all())

    async def count(
        self,
        name: str | None = None,
        age: int | None = None,
        status: str | None = None,
    ) -> int:
        stmt = select(func.count()).select_from(User)
        if name is not None:
            stmt = stmt.where(User.name == name)
        if age is not None:
            stmt = stmt.where(User.age == age)
        if status is not None:
            stmt = stmt.where(User.status == status)
        return await self.session.scalar(stmt) or 0

    async def save(self, user: User) -> User:
        self.session.add(user)
        await self.session.commit()
        return user

    async def delete(self, user: User) -> None:
        await self.session.delete(user)
        await self.session.commit()

    # ...and exists(), save_all(), delete_all(), find_all_paginated(),
    #    operator filters (__in / __ne / __like), row locking, soft delete —
    #    all written again for every entity.
```

**After** — `fast-repository` provides all of the above, pattern intact:

```python
from abc import ABC

from fast_repository import CRUDRepositoryInterface, CRUDRepository

# Domain layer: depend on this interface.
class UserRepositoryInterface(CRUDRepositoryInterface[User], ABC): ...

# Infrastructure layer: zero boilerplate, all CRUD methods provided.
class UserRepository(CRUDRepository[User], UserRepositoryInterface): ...
```

The entity class is captured from the generic argument (`CRUDRepository[User]`)
at class-definition time — no constructor wiring, no metaclass tricks to learn.

Using a synchronous `Session`? `SyncCRUDRepository` and
`SyncCRUDRepositoryInterface` offer the same API without `async`/`await`.

## Installation

```bash
pip install fast-repository
```

Requires Python 3.10+, SQLAlchemy 2.0+ (async or sync), and fastapi-pagination.

## Usage

```python
repo = UserRepository(session)  # AsyncSession

await repo.find(1)                                # SELECT ... WHERE id = 1
await repo.find(1, with_for_update=True)          # ... FOR UPDATE (row lock)
await repo.find(user_id=1, group_id=2)            # composite key by name
await repo.find_all(status="active")              # ... WHERE status = 'active'
await repo.find_all(id__in=[1, 2, 3])             # ... WHERE id IN (1, 2, 3)
await repo.find_all(age__ge=18, name__like="K%") # operator suffixes
await repo.find_all(or_(User.age < 18, User.age >= 65))  # raw SQLAlchemy expressions
await repo.find_all(status__ne="active")          # ... WHERE status != 'active'
await repo.find_all(id__notin=[1, 2, 3])          # ... WHERE id NOT IN (1, 2, 3)
await repo.find_all(order_by=User.age.desc())     # ... ORDER BY age DESC
await repo.find_all_paginated(params=Params(page=1, size=50), status="active")  # params optional under FastAPI
await repo.count(status="active")                 # SELECT count(*) ... WHERE status = 'active'
await repo.exists(id=1)                           # SELECT EXISTS(...) -> bool

await repo.save(user)
await repo.save_all(users)
await repo.delete(user)
await repo.delete_all(users)
```

### Filter syntax

| Keyword                                 | SQL                     |
|-----------------------------------------|-------------------------|
| `column=value`                          | `column = value`        |
| `column__in=[a, b]`                     | `column IN (a, b)`      |
| `column__notin=[a, b]`                  | `column NOT IN (a, b)`  |
| `column__ne=value`                      | `column != value`       |
| `column__gt` / `__ge` / `__lt` / `__le` | `>` / `>=` / `<` / `<=` |
| `column__like` / `__ilike`              | `LIKE` / `ILIKE`        |
| `column__is=None`                       | `IS NULL`               |

Unknown columns and operators raise `InvalidFilterError` instead of being
silently ignored, so a typo can never return unfiltered data.

### Customizing queries

Declare a base `stmt` to change relationship loading or apply a default
filter to every read. Pass it as a class keyword argument:

```python
class UserRepository(
    CRUDRepository[User],
    stmt=select(User).options(selectinload(User.posts)),
):
    ...
```

When omitted, reads default to `select(User)`.

### Typed filters for IDE autocomplete

`find_all`, `count`, and the other read methods take arbitrary keyword filters,
so a type checker can't suggest your entity's columns by name. To get IDE autocomplete,
re-declare the method on your interface like this:

```python
from sqlalchemy.sql import ColumnElement


class UserRepositoryInterface(CRUDRepositoryInterface[User], ABC):
    @abstractmethod
    async def find_all(
        self,
        *criteria: ColumnElement[bool],
        status: str | None = None,
        **_,
    ) -> list[User]: ...
```

Wherever a value is typed as `UserRepositoryInterface`, the editor now suggests `status=`.
Expose more filters by adding more keywords:

```python
class UserRepositoryInterface(CRUDRepositoryInterface[User], ABC):
    @abstractmethod
    async def find_all(
        self,
        *criteria: ColumnElement[bool],
        status: str | None = None,
        age: int | None = None,
        **_,
    ) -> list[User]: ...
```

This takes effect only when the variable is typed as the interface.

### Pagination with FastAPI

Inside a FastAPI route you do not need to pass `params` at all. When the
response model is a `Page[...]` and `add_pagination(app)` is wired up,
fastapi-pagination parses `?page=`/`?size=` from the query string and
`find_all_paginated` picks them up automatically:

```python
from fastapi import FastAPI
from fastapi_pagination import Page, add_pagination

app = FastAPI()

@app.get("/users", response_model=Page[UserOut])
async def list_users(repo: UserRepo, status: str | None = None) -> Page[User]:
    filters = {"status": status} if status is not None else {}
    return await repo.find_all_paginated(**filters)  # params injected from the request

add_pagination(app)
```

## Documentation

- [Getting Started](https://github.com/2u2s/fast-repository/blob/main/docs/en/getting-started.md) — install, define an entity, wire up a repository.
- [Filtering](https://github.com/2u2s/fast-repository/blob/main/docs/en/filtering.md) — keyword filters, operator suffixes, primary-key lookups, pagination.
- [Customizing queries](https://github.com/2u2s/fast-repository/blob/main/docs/en/customizing-queries.md) — eager-load relationships or apply a default filter to every read.
- [Transactions](https://github.com/2u2s/fast-repository/blob/main/docs/en/transactions.md) — control commit with the `autocommit` flag and group work into a unit of work.
- [Soft delete](https://github.com/2u2s/fast-repository/blob/main/docs/en/soft-delete.md) — opt in to marking rows deleted instead of removing them.
- [FastAPI integration](https://github.com/2u2s/fast-repository/blob/main/docs/en/fastapi.md) — wire the repository into routes with dependency injection.

Runnable [examples](https://github.com/2u2s/fast-repository/blob/main/examples/README.md) cover basic CRUD, filtering, the
`autocommit` flag, and a small FastAPI app.

## License

MIT
