Fat Service Layer, Brittle Tests — The Repository Pattern Is the Fix

1. The Problem

While building the backend for my personal project —I initially wrote database queries directly alongside the business rules. A function takes a database session, fetches a User record, updates an Book bookmark, commits the transaction, and returns the response.

But testing this became an architectural nightmare. The real problem is this: when business logic and database queries live in the same function, you literally cannot test the logic without hitting the database. There is no seam to inject anything. You cannot simply ask "does the toggle logic work correctly" without also asking "is the database running, is the schema correct, is the session valid". The logic and the I/O are completely fused. To test the logic, I was forced to deal with infrastructure. That is the exact problem the repository pattern solves.

2. What the Repository Pattern Is

The Repository Pattern is a design abstraction that separates your data access layer from your core business logic. Instead of controllers interacting with SQLAlchemy directly, they talk to a repository.

Router → Service → Repository → Database

The Router handles incoming HTTP requests and responses. The Service executes the business logic and rules. The Repository interacts with the database session to fetch or save records. The Database just stores the state.

3. The Before — How The Service Looked

Here is a snippet of how the bookmark toggling logic originally looked in my project. The service function handles both the database session and the business logic directly.

# controllers/bookmark_controller.py
async def toggle_bookmark_book(book_id: int, user_id: str, session: AsyncSession):
    # Fused DB Query: Hard to mock without replicating SQLAlchemy's API
    user = await session.get(User, user_id)
    if not user:
        raise ValueError("User not found")

    # Core Business Logic: Trapped inside the DB transaction
    books = list(user.favourite_books or [])
    action = book_id not in books
    
    books.append(book_id) if action else books.remove(book_id)
    user.favourite_books = books
    
    session.add(user)
    
    # Fused DB Command: Forces the test to handle transaction commits
    await session.commit()

    return {"bookmarked": action, "favourite_books": books}

Because the SQLAlchemy session is passed directly into the function, there is no boundary between data retrieval and data manipulation. You cannot easily mock the session to inject test data without replicating the entire ORM framework's behaviour. This fusion means isolating logic failures is hard, as any test failure could be a database error, a mock configuration issue, or a genuine bug in the business rules.

3b. What Testing Looks Like Before

Here is what writing a test for that fused function looks like without the repository pattern.

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_toggle_bookmark_book_ugly_mock():
    # Brittle: Mocking the SQLAlchemy implementation details
    mock_session = AsyncMock()
    mock_session.add = MagicMock()  # session.add is synchronous
    
    mock_user = AsyncMock()
    mock_user.favourite_books = [10]
    
    # Replicating the exact ORM API call just to test the array logic
    mock_session.get.return_value = mock_user

    result = await toggle_bookmark_book(42, "u1", mock_session)

    assert result["bookmarked"] is True
    assert result["favourite_books"] == [10, 42]
    
    # Verifying the exact ORM method was called
    mock_session.commit.assert_awaited_once()

This test is inherently brittle because it forces you to mock SQLAlchemy's internal implementation details rather than the application behaviour. If you refactor the internal database call from session.get() to session.execute(), this test will instantly break even if the output and behaviour remain perfectly identical.

4. The After — The Repository Pattern

To fix the fused logic, I extracted all SQLAlchemy queries into a dedicated BookmarkRepository using static methods. This encapsulates the database session handling away from the business logic.

# repositories/bookmark_repository.py
from core.database import engine
from sqlmodel import Session
from core.models import User

class BookmarkRepository:
    @staticmethod
    def toggle_book(user_id: str, book_id: int):
        with Session(engine) as session:
            user = session.get(User, user_id)
            if not user:
                return None
            
            books = list(user.favourite_books or [])
            action = book_id not in books
            
            books.append(book_id) if action else books.remove(book_id)
            
            user.favourite_books = books
            session.add(user)
            session.commit()
            return action, books

Now, the controller just orchestrates the logic, caching, and analytics without touching a database session.

# controllers/bookmark_controller.py
from repositories.bookmark_repository import BookmarkRepository
from fastapi import HTTPException
from core.redis import redis_client
from core.analytics import posthog

def toggle_bookmark_book(user_id: str, book_id: int):
    # 1. Ask Repo to handle the database transaction
    result = BookmarkRepository.toggle_book(user_id, book_id)
    if result is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    action, books = result
    
    # 2. Invalidate Cache
    redis_client.delete(f"bookmarks:books_data:{user_id}")
    
    # 3. Fire Analytics
    event_name = "book_bookmarked" if action else "book_unbookmarked"
    posthog.capture(distinct_id=user_id, event=event_name)
    
    return {"bookmarked": action, "favourite_books": books}

By doing this, the controller knows nothing about the database. Because we used static methods, we can easily swap the real PostgreSQL repository for a mocked one during tests using Python's patch. The business rules and side effects (caching, analytics) are isolated from the database interactions.

5. What Testing Looks Like Now

5a. Testing the Controller (pure logic, zero I/O)

import pytest
from unittest.mock import patch
import controllers.bookmark_controller as bookmarks

@patch("controllers.bookmark_controller.BookmarkRepository")
def test_toggle_bookmark_book_adds_book(mock_repo, module_deps):
    # Arrange: Fake Redis and Analytics injected via module_deps
    redis, posthog, _ = module_deps
    
    # Intercept the static DB call and force a successful return
    mock_repo.toggle_book.return_value = (True, [1, 42])
    redis.set("bookmarks:books_data:u1", "old_cache")

    # Act: Execute pure business logic
    result = bookmarks.toggle_bookmark_book("u1", 42)

    # Assert: Verify state mutation, cache invalidation, and analytics
    assert result == {"bookmarked": True, "favourite_books": [1, 42]}
    assert redis.exists("bookmarks:books_data:u1") == 0  # Cache cleared!
    
    posthog.capture.assert_called_once()

This test has zero database I/O and runs in a fraction of a millisecond. Because we mock the repository, we can test exactly how the controller handles caching and analytics without worrying about database setup or teardown.

5b. Testing the Repository (real database, isolated concern)

import pytest
from sqlmodel import Session
from core.models import User
from repositories.bookmark_repository import BookmarkRepository

def test_repository_persists_user_bookmark_state(db_session: Session):
    # Arrange: Seed real database state using the Testcontainers session
    test_user = User(id="user_123", favourite_books=[])
    db_session.add(test_user)
    db_session.commit()

    # Act: Test the static repository method
    action, updated_books = BookmarkRepository.toggle_book("user_123", 99)

    # Assert: Verify SQL execution against the real schema
    assert action is True
    assert updated_books == [99]
    
    # Verify persistence directly from the DB
    fetched_user = db_session.get(User, "user_123")
    assert fetched_user.favourite_books == [99]

These tests ensure your ORM configs and raw SQL work correctly against a real PostgreSQL schema, isolating database checks from application logic. These two test types serve completely different purposes and should never be collapsed into one.

6. The Folder Structure

app/
  routers/
  services/
  repositories/
  models/
  schemas/

This is not bureaucracy. It is the minimum structure that makes each layer independently testable.

7. Closing

The pattern costs you one extra file per domain. What it buys is a service layer you can test without a database, a repository you can test without business logic, and a router that has no opinion about either. When a test fails, you know exactly which layer broke and why. That is the only thing a test suite needs to guarantee.