Implementing the Repository Pattern

The Problem: Tightly Coupled Code

Imagine you're writing a path operation to create a new user. Without a clear separation of concerns, your route handler might look something like this:

@router.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):



    # 1. Check if user exists (Database logic)
    query = select(User).where(User.email == user.email)
    result = await db.execute(query)
    existing_user = result.scalars().first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered.")




    # 2. Hash the password (Business logic)
    hashed_password = get_password_hash(user.password)
    



    # 3. Create the user object (Database logic)
    db_user = User(email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    
    return db_user

As the code from the "FastAPI: SOLID Principles and Design Patterns" article highlights, this endpoint mixes responsibilities: HTTP request/response handling, business validation, and direct database queries.

This tight coupling makes the code difficult to test (you always need a database connection) and hard to maintain. If you change your database schema or switch ORMs, you'd have to hunt down and update every route that touches the database.

The Solution: The Repository Pattern

The repository pattern solves this by creating an abstraction layer that sits between your business logic and your data persistence mechanism (e.g., SQLAlchemy). It acts as an intermediary, exposing a simple, collection-like interface for accessing your data.

Your business logic will now ask the repository, "give me a user by this ID" or "save this new user," without needing to know anything about database sessions, SQL queries, or ORM objects.

Key benefits include:

• Decoupling: Your application's core logic is independent of the database technology.
• Testability: You can easily "mock" the repository in your tests with an in-memory version, making your tests faster and more reliable.
• Reusability: All queries related to a specific model are centralized in one place.

Building Your First Repository

Let's watch a video that provides an excellent, step-by-step guide to refactoring a tightly coupled API to use the repository pattern.

Let's summarize the key steps from the video with code.

Step 1: Define the Repository Class

First, you create a new file, for example, app/repositories/user_repository.py. This class will encapsulate all database operations for the User model. A crucial design choice is to pass the database session in the constructor. This is fundamental for dependency injection.

# app/repositories/user_repository.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.user import User  # Your SQLAlchemy model
from app.schemas.user import UserCreate # Your Pydantic schema

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_user_by_email(self, email: str) -> User | None:
        query = select(User).where(User.email == email)
        result = await self.db.execute(query)
        return result.scalars().first()

    async def create_user(self, user_data: UserCreate, hashed_password: str) -> User:
        db_user = User(email=user_data.email, hashed_password=hashed_password)
        self.db.add(db_user)
        await self.db.commit()
        await self.db.refresh(db_user)
        return db_user

Step 2: Refactor the Route to Use the Repository

Now, your create_user path operation becomes much cleaner. It no longer contains any direct database code. Instead, it delegates that responsibility to the repository.

To make this work, we use FastAPI's Depends to inject an instance of our UserRepository.

# app/routers/users.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_db, get_user_repository # We'll create get_user_repository next
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate, UserRead

router = APIRouter()

@router.post("/users", response_model=UserRead)
async def create_user(
    user: UserCreate, 
    repo: UserRepository = Depends(get_user_repository)
):



    # Business logic now lives here or in a service layer
    db_user = await repo.get_user_by_email(email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered.")
    



    # Let's assume password hashing is handled in a service (next lesson!)
    # For now, we pass it to the repository
    hashed_password = "a_real_hashed_password" # Placeholder
    
    return await repo.create_user(user_data=user, hashed_password=hashed_password)

Step 3: Create the Dependency

In your app/dependencies.py file, you create a function that knows how to provide a UserRepository instance. This function itself depends on getting a database session.

# app/dependencies.py

from sqlalchemy.ext.asyncio import AsyncSession
from app.database import SessionLocal # Assuming you have this
from app.repositories.user_repository import UserRepository




# Dependency to get a DB session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session




# Dependency to get the user repository
def get_user_repository(db: AsyncSession = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

This chain of dependencies (get_user_repository depends on get_db) is managed automatically by FastAPI.

The Payoff: Clean, Isolated Testing

The true power of this pattern becomes evident when you write tests. You can now test your create_user endpoint without ever touching a real database.

The video you watched introduces the idea of using a "dummy" repository for testing. This is a crucial concept.

To test our user creation endpoint, you would create an InMemoryUserRepository that mimics the real one but uses a simple dictionary.

# tests/repositories/in_memory_user_repo.py

class InMemoryUserRepository:
    def __init__(self):
        self._users = {}
        self._next_id = 1

    async def get_user_by_email(self, email: str):
        for user in self._users.values():
            if user.email == email:
                return user
        return None

    async def create_user(self, user_data, hashed_password):
        new_user_id = self._next_id
        new_user = User(id=new_user_id, email=user_data.email, hashed_password=hashed_password)
        self._users[new_user_id] = new_user
        self._next_id += 1
        return new_user




# In your test file (e.g., tests/test_users.py)
from app.main import app
from app.dependencies import get_user_repository
from fastapi.testclient import TestClient

client = TestClient(app)

def override_get_user_repository():
    return InMemoryUserRepository()




# Override the dependency for all tests in this module
app.dependency_overrides[get_user_repository] = override_get_user_repository

def test_create_user_success():
    response = client.post(
        "/users",
        json={"email": "test@example.com", "password": "password123"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "id" in data

This is FastAPI's dependency_overrides in action. It's a clean, powerful way to swap out dependencies for testing, and it's the direct result of structuring your application with patterns like the repository.

Connecting to Your Django Experience

This pattern brings the kind of structured data access that is often manually implemented in larger Django projects into a first-class, framework-supported pattern in FastAPI.

Think of it this way:

• In a simple Django view, you might write Book.objects.get(pk=book_id).
• In a more complex application, to avoid repeating logic, you might create a books/services.py or books/queries.py file with a function get_book(book_id).
• The repository pattern formalizes this by creating a BookRepository class with a get(book_id) method, which is then injected as a dependency.

This formalization brings the significant benefits of explicit dependencies and easy testability that you've seen today.

Conclusion

You have now learned one of the most important patterns for building robust and scalable APIs. By isolating data access logic into a repository, you've made your application cleaner, more maintainable, and significantly easier to test.

Key Takeaways:

• The Repository Pattern's Purpose: To create an abstraction layer that decouples your business logic from the data persistence layer.
• Implementation: A repository is typically a class that takes a database session as a parameter and exposes methods for CRUD operations (e.g., get_by_id, create, list).
• Integration with FastAPI: Use FastAPI's dependency injection system to provide repository instances to your path operations. This involves creating a dependency function that instantiates the repository with a database session.
• The Main Benefit is Testability: This pattern allows you to use dependency_overrides to swap in a fake, in-memory repository during testing, enabling fast, isolated unit tests for your API endpoints.

Preview of the Next Lesson:

Our path operations are now much cleaner, but they still contain some business logic (like checking if a user exists). To achieve an even cleaner separation of concerns, we need another layer. In the next lesson, we will implement a service layer to encapsulate business logic, which will orchestrate calls between route handlers and the repositories we just built.