fast-repository 0.2.1

Interface-first repository pattern for FastAPI + SQLAlchemy. Declare the interface, get the implementation for free.

fast-repository

English | 한국어

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:

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:

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

pip install fast-repository

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

Usage

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:

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:

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:

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:

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 — install, define an entity, wire up a repository.
• Filtering — keyword filters, operator suffixes, primary-key lookups, pagination.
• Customizing queries — eager-load relationships or apply a default filter to every read.
• Transactions — control commit with the autocommit flag and group work into a unit of work.
• Soft delete — opt in to marking rows deleted instead of removing them.
• FastAPI integration — wire the repository into routes with dependency injection.

Runnable examples cover basic CRUD, filtering, the autocommit flag, and a small FastAPI app.

License

MIT