Build Once, Reuse Anywhere: Generic Repository Pattern in Python

Problem: Repeated CRUD code

In many FastAPI/SQLAlchemy projects each entity has its own repository (UserRepository, ProductRepository, OrderRepository) that repeats the same save, get, update, delete, pagination and error‑handling logic. The author shows an example where the same code is copied eight times and points out the risk of bugs when business logic changes.

Solution: Generic abstract repository

Introduce a generic abstract base class that implements all common CRUD operations, pagination, sorting, existence check and counting, while remaining type‑safe through Python generics. The base class defines abstract methods _model_to_entity, _entity_to_model and _get_filters that concrete repositories must implement.

Core components

Entity base class ( EntityBase) with id, created_at, updated_at.

Utility types: DatabaseException, Ordering enum for asc / desc.

Abstract repository SqlAlchemyAbstractRepository[Entity, SqlAlchemyModel] with methods save, update, list_all, get, exists, delete, count.

Concrete implementation example

For a User entity the concrete repository inherits from the abstract class and supplies the ORM model and the mapping logic.

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):
        """Support id, email, role 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

Why _get_filters matters

All query APIs delegate filter construction to this single method, keeping the public API clean and flexible. Example calls:

# List admins
admins = await user_repo.list_all(role_filter="admin", page=1, limit=20)

# Find a user by email
user = await user_repo.get(email_filter="[email protected]")

# Check existence
exists = await user_repo.exists(email_filter="[email protected]")

Custom error handling

Subclasses can override save to translate database‑specific errors, e.g., catching IntegrityError for duplicate email and raising a domain‑specific UserAlreadyExistsError. Forgetting to call await self._session.rollback() leaves the session in an error state – a common pitfall.

Extending the generic repository

Additional domain‑specific methods can be added without breaking the generic base, such as get_by_email or get_active_admins, which internally reuse the generic get and list_all methods.

Real‑world impact

After refactoring a medium‑size FastAPI project, the author measured:

Number of repositories unchanged (8).

Lines per repository reduced from 250‑400 to 30‑50.

All duplicated CRUD code eliminated.

Pagination logic changed in one place instead of eight.

Type safety gained (compile‑time checks).

The consolidation reduced bug surface and improved maintainability.

Why adopt this pattern

DRY principle – write once, modify once.

Consistent behavior across repositories, low onboarding cost.

Static type checking prevents accidental misuse.

Test the abstract base once for full coverage.

Easy to add cross‑cutting features (soft delete, optimistic lock) by changing the base.

Flexibility – override methods when special behavior is needed.

Conclusion

Using a generic, abstract repository with async SQLAlchemy brings a qualitative jump in code quality for Python data‑access layers. New entities require only a few lines of mapping and filter definitions, while the heavy lifting of CRUD, pagination, sorting and error handling is reused automatically.