How a Python Generic Repository Cuts 80% of Duplicate CRUD Code

Problem

FastAPI projects often contain separate repository classes (e.g., UserRepository, ProductRepository, OrderRepository) that duplicate the same CRUD, pagination, error‑handling, and SQLAlchemy query logic. The only differences are the entity type, the ORM model type, and the mapping between them, leading to a “copy‑paste syndrome” and bugs when changes must be applied in many places.

Solution

A single generic abstract repository, built with Python generics, captures the entity and model types and provides all common data‑access operations. Concrete repositories only need to specify the ORM model class and implement three abstract methods: mapping from entity to model, mapping from model to entity, and translating filter arguments into SQLAlchemy expressions.

Core components

Entity base class (type‑safe, timestamped):

from dataclasses import dataclass, field
from datetime import datetime, timezone
from uuid import UUID

@dataclass(kw_only=True)
class EntityBase:
    id: UUID | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

Utility types for error handling and ordering:

class DatabaseException(Exception):
    """Unified wrapper for database operation errors"""
    pass

from enum import StrEnum

class Ordering(StrEnum):
    """Type‑safe sorting direction"""
    asc = "asc"
    desc = "desc"

Abstract generic repository (simplified)

from abc import ABC, abstractmethod
from typing import Any, Generic, List, TypeVar
import sqlalchemy
from sqlalchemy import asc, desc, func, select
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

Entity = TypeVar("Entity", bound=EntityBase)
SqlAlchemyModel = TypeVar("SqlAlchemyModel", bound=Base)

class SqlAlchemyAbstractRepository(ABC, Generic[Entity, SqlAlchemyModel]):
    model: type[SqlAlchemyModel]

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

    async def save(self, entity: Entity) -> Entity:
        model = self._entity_to_model(entity)
        self._session.add(model)
        await self._session.flush()
        await self._session.refresh(model)
        return self._model_to_entity(model)

    async def list_all(self, page: int = 1, limit: int = 10,
                       order_by: str = "created_at",
                       ordering: Ordering = Ordering.asc,
                       **filters) -> List[Entity]:
        query = select(self.model)
        query = query.where(*self._get_filters(**filters))
        query = query.order_by(self._get_order_expression(order_by, ordering))
        offset = (page - 1) * limit
        query = query.offset(offset).limit(limit)
        result = await self._session.execute(query)
        models = result.scalars().all()
        return [self._model_to_entity(m) for m in models]

    async def get(self, **filters) -> Entity | None:
        query = select(self.model).where(*self._get_filters(**filters))
        model = await self._session.scalar(query)
        return self._model_to_entity(model) if model else None

    async def exists(self, **filters) -> bool:
        query = select(self.model).where(*self._get_filters(**filters))
        result = await self._session.scalar(query)
        return result is not None

    async def delete(self, **filters) -> int:
        try:
            query = sqlalchemy.delete(self.model).where(*self._get_filters(**filters))
            result = await self._session.execute(query)
            await self._session.flush()
            return result.rowcount  # type: ignore[attr-defined]
        except SQLAlchemyError as e:
            await self._session.rollback()
            raise DatabaseException from e

    async def count(self, **filters) -> int:
        filter_conditions = self._get_filters(**filters)
        return await self._session.scalar(
            select(func.count()).select_from(self.model).where(*filter_conditions)
        ) or 0

    @staticmethod
    @abstractmethod
    def _model_to_entity(model: SqlAlchemyModel) -> Entity:
        raise NotImplementedError

    @staticmethod
    @abstractmethod
    def _entity_to_model(entity: Entity) -> SqlAlchemyModel:
        raise NotImplementedError

    def _get_filters(self, **filters) -> List[Any]:
        return []

    @staticmethod
    def _get_order_expression(order_by: str, ordering: Ordering):
        return asc(order_by) if ordering == Ordering.asc else desc(order_by)

Concrete repository example for User

class SqlAlchemyUserRepository(SqlAlchemyAbstractRepository[User, UserModel]):
    model = UserModel

    def _entity_to_model(self, entity: User) -> UserModel:
        model = UserModel(name=entity.name, email=entity.email, role=entity.role)
        if entity.id:
            model.id = entity.id
        return model

    def _model_to_entity(self, model: UserModel) -> User:
        return User(
            id=model.id,
            name=model.name,
            email=model.email,
            role=model.role,
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    def _get_filters(self, **filters):
        conditions = []
        if "id_filter" in filters:
            conditions.append(UserModel.id == filters["id_filter"])
        if "email_filter" in filters:
            conditions.append(UserModel.email == filters["email_filter"])
        if "role_filter" in filters:
            conditions.append(UserModel.role == filters["role_filter"])
        return conditions

Key point: forgetting await self._session.rollback() after an exception leaves the session in a broken state.

Custom query methods can be added without breaking the generic contract:

async def get_by_email(self, email: str) -> User | None:
    return await self.get(email_filter=email)

async def get_active_admins(self) -> List[User]:
    return await self.list_all(role_filter="admin", status_filter="active")

Real‑world impact (before vs. after refactor)

Repository count unchanged (8), but each file shrank from 250‑400 lines to 30‑50 lines.

Duplicate CRUD code eliminated.

Pagination logic changes reduced from 8 locations to a single location.

Type safety improved from unchecked parameters to compile‑time checks.

Why adopt this pattern?

DRY principle : write once, fix once, all repositories benefit.

Consistency : uniform behavior lowers onboarding friction.

Type safety : static checks prevent accidental misuse.

Testability : unit‑test the abstract base once for all concrete repositories.

Maintainability : global changes (e.g., soft‑delete) require only a single edit.

Flexibility : custom behavior is added by overriding methods; the base does not restrict extensions.

Typical migration strategy: start with a non‑critical repository, verify correctness, then incrementally replace the remaining repositories.