KAHIBARO
Discord Login Register ☰
☰

12.15. Repository Pattern

Table of Contents

Why Use the Repository Pattern?

In a typical backend application you will write code that:

If you mix these in the same place, you quickly end up with code that is:

The Repository Pattern solves this by putting all data access behind a clear interface. Your application code speaks to repositories, not directly to the ORM.

A simple mental picture:

Your business logic only depends on interfaces, which makes it easier to test and maintain.

Core Ideas of the Repository Pattern

Repository as a Collection-like Interface

A repository represents a collection of objects of a certain type, for example users or orders. Think of it like a specialized in-memory list, but backed by a database.

Typical operations:

The important part is that your application does not care how these operations are implemented. It just calls methods on a repository.

A very small example in Python-style pseudocode:

python 
class User:
    id: int
    email: str
    name: str
class UserRepository:
    def add(self, user: User) -> User:
        ...
    def get(self, user_id: int) -> User | None:
        ...
    def list(self) -> list[User]:
        ...
    def find_by_email(self, email: str) -> User | None:
        ...

This is just an interface or abstract idea. The real work happens in concrete implementations.

Separating Interface from Implementation

The Repository Pattern is about abstraction. You define:

  1. A repository interface that describes what can be done
  2. One or more implementations that describe how it is done

You might have:

The interface stays the same, only the implementation changes.

In Python, this often looks like:

python 
from abc import ABC, abstractmethod
class UserRepository(ABC):
    @abstractmethod
    def add(self, user: User) -> User:
        ...
    @abstractmethod
    def get(self, user_id: int) -> User | None:
        ...
    @abstractmethod
    def find_by_email(self, email: str) -> User | None:
        ...
class SqlAlchemyUserRepository(UserRepository):
    def __init__(self, session):
        self.session = session
    def add(self, user: User) -> User:
        self.session.add(user)
        self.session.flush()
        return user
    def get(self, user_id: int) -> User | None:
        return self.session.get(User, user_id)
    def find_by_email(self, email: str) -> User | None:
        return (
            self.session.query(User)
            .filter(User.email == email)
            .one_or_none()
        )

Your route handlers and services only know about UserRepository, not about SqlAlchemyUserRepository.

Benefits in a Backend Application

Decoupling Business Logic from Data Access

When you do not use repositories, you often see code like this in a FastAPI endpoint:

python 
def create_user(db_session: Session, user_in: UserCreate):
    # ORM code and business logic mixed
    existing = (
        db_session.query(User)
        .filter(User.email == user_in.email)
        .one_or_none()
    )
    if existing:
        raise HTTPException(status_code=400, detail="Email already used")
    user = User(
        email=user_in.email,
        name=user_in.name,
    )
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)
    return user

The endpoint:

With a repository, you can separate roles:

python 
class UserService:
    def __init__(self, users: UserRepository):
        self.users = users
    def register_user(self, email: str, name: str) -> User:
        if self.users.find_by_email(email):
            raise EmailAlreadyUsed()
        user = User(email=email, name=name)
        self.users.add(user)
        return user

Then your FastAPI endpoint becomes:

python 
def create_user(
    user_in: UserCreate,
    service: UserService = Depends(get_user_service),
):
    try:
        user = service.register_user(
            email=user_in.email,
            name=user_in.name,
        )
    except EmailAlreadyUsed:
        raise HTTPException(status_code=400, detail="Email already used")
    return user

Now:

Easier Unit Testing

With a repository interface, you can swap the database implementation for something simpler in tests.

For example, an in-memory repository:

python 
class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._items: dict[int, User] = {}
        self._id_counter = 1
    def add(self, user: User) -> User:
        if getattr(user, "id", None) is None:
            user.id = self._id_counter
            self._id_counter += 1
        self._items[user.id] = user
        return user
    def get(self, user_id: int) -> User | None:
        return self._items.get(user_id)
    def find_by_email(self, email: str) -> User | None:
        return next(
            (u for u in self._items.values() if u.email == email),
            None,
        )

Then a unit test for business logic does not need a database at all:

python 
def test_register_user_creates_new_user():
    repo = InMemoryUserRepository()
    service = UserService(users=repo)
    user = service.register_user(
        email="test@example.com",
        name="Test User",
    )
    assert user.id is not None
    assert repo.get(user.id).email == "test@example.com"

You test the logic without worrying about:

This usually makes tests much faster and simpler.

Supporting Multiple Storage Technologies

Because the application only depends on the interface, you can:

For example, two implementations of the same repository:

python 
class SqlAlchemyOrderRepository(OrderRepository):
    # uses PostgreSQL via SQLAlchemy
    ...
class RedisCachedOrderRepository(OrderRepository):
    def __init__(self, inner: OrderRepository, redis_client):
        self.inner = inner
        self.redis = redis_client
    def get(self, order_id: int) -> Order | None:
        cache_key = f"order:{order_id}"
        cached = self.redis.get(cache_key)
        if cached:
            return deserialize_order(cached)
        order = self.inner.get(order_id)
        if order:
            self.redis.set(cache_key, serialize_order(order))
        return order
    def add(self, order: Order) -> Order:
        order = self.inner.add(order)
        cache_key = f"order:{order.id}"
        self.redis.set(cache_key, serialize_order(order))
        return order

The service layer does not change. Only the wiring changes when you decide which concrete repository to use.

Typical Repository Design in Python with SQLAlchemy

Below is a small, concrete example that mirrors a real project structure. Details of sessions, models, and transactions belong to other chapters, so we keep those parts simple.

Assume you have a User ORM model.

python 
# models.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
    pass
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    name: Mapped[str]

Now define a repository interface and its SQLAlchemy implementation.

python 
# repositories.py
from abc import ABC, abstractmethod
from sqlalchemy.orm import Session
from .models import User
class UserRepository(ABC):
    @abstractmethod
    def add(self, user: User) -> User:
        ...
    @abstractmethod
    def get(self, user_id: int) -> User | None:
        ...
    @abstractmethod
    def list(self) -> list[User]:
        ...
    @abstractmethod
    def find_by_email(self, email: str) -> User | None:
        ...
class SqlAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session
    def add(self, user: User) -> User:
        self.session.add(user)
        # flush to get generated primary key without committing
        self.session.flush()
        return user
    def get(self, user_id: int) -> User | None:
        return self.session.get(User, user_id)
    def list(self) -> list[User]:
        return self.session.query(User).all()
    def find_by_email(self, email: str) -> User | None:
        return (
            self.session.query(User)
            .filter(User.email == email)
            .one_or_none()
        )

Your service layer consumes the interface:

python 
# services.py
class UserService:
    def __init__(self, users: UserRepository):
        self.users = users
    def register_user(self, email: str, name: str) -> User:
        if self.users.find_by_email(email):
            raise EmailAlreadyUsed()
        user = User(email=email, name=name)
        self.users.add(user)
        return user

Then in your FastAPI dependency injection, you bind the interface to an implementation:

python 
# dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session
from .db import get_session
from .repositories import SqlAlchemyUserRepository
from .services import UserService
def get_user_repository(
    session: Session = Depends(get_session),
) -> SqlAlchemyUserRepository:
    return SqlAlchemyUserRepository(session=session)
def get_user_service(
    repo = Depends(get_user_repository),
) -> UserService:
    return UserService(users=repo)

When you import get_user_service into routes, FastAPI will create SqlAlchemyUserRepository instances for each request automatically.

Common Variations of the Repository Pattern

One Repository per Aggregate or Entity

In many systems you will have:

Each repository handles its own entity type. Methods in each repository are specific and meaningful to that entity.

Example repository methods:

EntityExample methods
Userfind_by_email, list_active, search
Orderlist_by_user, list_pending, find_paid
Productsearch_by_name, list_in_category

The names express intent, not implementation details like join or filter.

Generic Base Repository

Some teams create a base repository with generic CRUD methods and then inherit from it.

python 
from typing import Generic, TypeVar, Type
from sqlalchemy.orm import Session
from .models import Base
T = TypeVar("T", bound=Base)
class SqlAlchemyRepository(Generic[T]):
    model_class: Type[T]
    def __init__(self, session: Session):
        self.session = session
    def add(self, entity: T) -> T:
        self.session.add(entity)
        self.session.flush()
        return entity
    def get(self, entity_id: int) -> T | None:
        return self.session.get(self.model_class, entity_id)
    def list(self) -> list[T]:
        return self.session.query(self.model_class).all()
class UserRepository(SqlAlchemyRepository[User]):
    model_class = User
    def find_by_email(self, email: str) -> User | None:
        return (
            self.session.query(User)
            .filter(User.email == email)
            .one_or_none()
        )

This reduces repetition for basic operations and lets you add entity-specific queries in each subclass.

Read and Write Separation

Sometimes you may have:

For example:

python 
class UserReadRepository(ABC):
    @abstractmethod
    def get(self, user_id: int) -> User | None:
        ...
    @abstractmethod
    def list(self) -> list[User]:
        ...
class UserWriteRepository(ABC):
    @abstractmethod
    def add(self, user: User) -> User:
        ...
    @abstractmethod
    def delete(self, user: User) -> None:
        ...

This pattern is useful for complex systems or CQRS architectures, but for beginner projects a single repository per entity is usually enough.

Practical Examples of Using Repositories

Example: Creating and Fetching a Product

Define a product repository:

python 
class ProductRepository(ABC):
    @abstractmethod
    def add(self, product: Product) -> Product:
        ...
    @abstractmethod
    def get(self, product_id: int) -> Product | None:
        ...
    @abstractmethod
    def list_in_stock(self) -> list[Product]:
        ...
class SqlAlchemyProductRepository(ProductRepository):
    def __init__(self, session: Session):
        self.session = session
    def add(self, product: Product) -> Product:
        self.session.add(product)
        self.session.flush()
        return product
    def get(self, product_id: int) -> Product | None:
        return self.session.get(Product, product_id)
    def list_in_stock(self) -> list[Product]:
        return (
            self.session.query(Product)
            .filter(Product.quantity > 0)
            .all()
        )

In a service:

python 
class ProductService:
    def __init__(self, products: ProductRepository):
        self.products = products
    def create_product(self, name: str, price: float, quantity: int) -> Product:
        product = Product(
            name=name,
            price=price,
            quantity=quantity,
        )
        self.products.add(product)
        return product
    def list_available_products(self) -> list[Product]:
        return self.products.list_in_stock()

In a FastAPI route:

python 
@router.post("/products")
def create_product(
    data: ProductCreate,
    service: ProductService = Depends(get_product_service),
):
    product = service.create_product(
        name=data.name,
        price=data.price,
        quantity=data.quantity,
    )
    return product
@router.get("/products")
def list_products(
    service: ProductService = Depends(get_product_service),
):
    return service.list_available_products()

You can now unit test ProductService with an in-memory ProductRepository without any web framework or database.

Transaction Handling and Repositories

Repositories often interact with database transactions. There are two common approaches:

  1. The unit of work pattern handles the transaction and passes a session into repositories.
  2. The repository itself begins and commits or rolls back transactions.

In modern applications, the first approach is more common. The repository just uses the session it receives. A higher-level component controls when to commit.

Example with a simple unit of work:

python 
class UnitOfWork(ABC):
    users: UserRepository
    @abstractmethod
    def commit(self) -> None:
        ...
    @abstractmethod
    def rollback(self) -> None:
        ...
class SqlAlchemyUnitOfWork(UnitOfWork):
    def __init__(self, session_factory):
        self.session_factory = session_factory
    def __enter__(self):
        self.session = self.session_factory()
        self.users = SqlAlchemyUserRepository(self.session)
        return self
    def __exit__(self, exc_type, exc, tb):
        if exc:
            self.rollback()
        else:
            self.commit()
        self.session.close()
    def commit(self) -> None:
        self.session.commit()
    def rollback(self) -> None:
        self.session.rollback()

Usage in a service:

python 
def register_user(uow: UnitOfWork, email: str, name: str) -> User:
    with uow:
        if uow.users.find_by_email(email):
            raise EmailAlreadyUsed()
        user = User(email=email, name=name)
        uow.users.add(user)
        return user  # committed on successful exit

Your business logic now uses repositories and unit of work instead of dealing with raw sessions.

When to Use the Repository Pattern

The Repository Pattern is most useful when:

For very small scripts or simple CRUD prototypes, a full repository layer might feel like extra work. But as projects grow, this separation pays off in clarity and testability.

Key Rules to Remember

Important rules for the Repository Pattern:

  1. Your business logic should depend on repository interfaces, not concrete ORM implementations.
  2. Repositories represent collections of aggregate roots or entities and expose intent-based methods like find_by_email, not raw SQL.
  3. Repository interfaces stay stable while implementations can change to use different databases, caching, or mocking for tests.
  4. Keep transaction control outside repositories, usually in a unit of work or service layer, so repositories focus on data access, not coordination.
  5. Tests for business logic should use in-memory or fake repository implementations, not the real database.

If you keep these rules in mind, repositories will help you build backend applications that are easier to change, easier to test, and easier to reason about as they grow.

Views: 9

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3706 / 24h
Impressum / Imprint
Datenschutz / Data Policy