Repository Pattern in FastAPI: Quando e Perché Usarlo

Separare business logic da data access: la guida che avrei voluto leggere prima di iniziare

---

⏱️ 12 minuti di lettura | 💻 Codice su GitHub → | 🎯 Livello: Intermedio

---

In Sintesi

Cos’è: Pattern architetturale che separa l’accesso al database dalla business logic

Problema che risolve:

• Business logic mischiata a query SQL
• Test lenti che richiedono setup database completo
• Difficoltà nel modificare query senza toccare logica

Quando usarlo:

• ✅ Business logic complessa
• ✅ Necessità di testare senza database
• ✅ Previsione di cambio database futuro
• ✅ Team in crescita

Quando NON usarlo:

• ❌ Prototipo veloce
• ❌ CRUD semplicissimo
• ❌ Progetto personale piccolo

Pattern:

Router → Service (business logic) → Repository (SQL) → Database

Implementazione | Quando Serve | Errori Comuni

---

Il Problema: Business Logic Mischiata a SQL

In applicazioni FastAPI senza un layer di astrazione, i router tendono a diventare così:

@router.post("/stories")
async def create_story(
    request: StoryCreate,
    db: AsyncSession = Depends(get_db)
):
    # Validation
    if not request.title or len(request.title) > 200:
        raise HTTPException(400, "Invalid title")

    # Direct SQL query
    story = Story(
        user_id=request.user_id,
        title=request.title,
        emotional_state=request.emotional_state
    )
    db.add(story)

    # Business logic (emotional score calculation)
    base_score = calculate_emotion_intensity(request.emotional_state)

    # Another query
    if request.previous_story_id:
        prev = await db.execute(
            select(Story).where(Story.id == request.previous_story_id)
        )
        prev_story = prev.scalar_one()
        progression = calculate_progression(prev_story, story)
        base_score *= progression

    story.emotional_score = base_score

    # Update user stats
    user = await db.execute(select(User).where(User.id == request.user_id))
    user = user.scalar_one()
    user.total_stories += 1
    user.emotional_score += base_score
    
    await db.commit()
    return story

Questo approccio funziona, ma ha problemi:

Problema 1: Test Complessi

async def test_create_story():
    # Requires full database setup
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session = AsyncSession(engine)

    # Setup test data
    user = User(id=uuid4(), username="test")
    session.add(user)
    await session.commit()

    # Run the actual test
    response = await create_story(...)

Il test funziona, ma richiede setup database completo per testare business logic.

Problema 2: Duplicazione

Quando aggiungo endpoint per “aggiorna storia”, devo copiare la logica di calcolo punteggio emozionale.

Bug nel calcolo? Va fixato in più posti.

Problema 3: Difficoltà Modifica

Cambiare da SQLite a PostgreSQL significa toccare tutti i file dove ci sono query dirette.

La Soluzione: Separazione delle Responsabilità

Il Repository Pattern introduce una separazione netta:

┌─────────────────────────────────────┐
│  ROUTER                             │
│  - HTTP handling                    │
│  - Input validation (Pydantic)      │
└──────────────┬──────────────────────┘
               │
               ↓
┌─────────────────────────────────────┐
│  SERVICE                            │
│  - Business logic                   │
│  - Orchestrazione                   │
│  - Calcoli                          │
└──────────────┬──────────────────────┘
               │
               ↓
┌─────────────────────────────────────┐
│  REPOSITORY                         │
│  - Query SQL                        │
│  - CRUD operations                  │
└──────────────┬──────────────────────┘
               │
               ↓
          [Database]

Stesso Codice Refactorato

Router (solo HTTP):

@router.post("/stories")
async def create_story(
    request: StoryCreate,
    story_service: StoryServiceDep
):
    return await story_service.create_story(request)

Service (business logic):

class StoryService:
    async def create_story(self, request: StoryCreate) -> Story:
        # Emotional score calculation (business logic)
        score = self._calculate_emotional_score(
            request.emotional_state,
            request.previous_story_id
        )

        # Create via repository
        story = await self.story_repo.create(
            Story(
                title=request.title,
                emotional_state=request.emotional_state,
                emotional_score=score
            )
        )

        # Update user stats via repository
        await self.user_repo.update_stats(request.user_id, score)
        
        return story

Repository (solo SQL):

class StoryRepository:
    async def create(self, story: Story) -> Story:
        self.db.add(story)
        await self.db.flush()
        await self.db.refresh(story)
        return story

Implementazione Completa

Step 1: BaseRepository Generico

# app/repositories/base_repository.py

from typing import TypeVar, Generic, Type, Optional, List
from uuid import UUID
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

T = TypeVar("T")

class BaseRepository(Generic[T]):
    """Repository base con operazioni CRUD comuni."""
    
    def __init__(self, session: AsyncSession, model: Type[T]):
        self.session = session
        self.model = model
    
    async def get(self, id: UUID) -> Optional[T]:
        """Get entity by ID."""
        result = await self.session.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()
    
    async def list(
        self,
        limit: int = 100,
        offset: int = 0,
        **filters
    ) -> List[T]:
        """List entities with filters."""
        query = select(self.model)
        
        for key, value in filters.items():
            if hasattr(self.model, key):
                query = query.where(getattr(self.model, key) == value)
        
        query = query.limit(limit).offset(offset)
        result = await self.session.execute(query)
        return list(result.scalars().all())
    
    async def create(self, obj: T) -> T:
        """Create new entity."""
        self.session.add(obj)
        await self.session.flush()
        await self.session.refresh(obj)
        return obj
    
    async def update(self, id: UUID, updates: dict) -> Optional[T]:
        """Update entity."""
        await self.session.execute(
            update(self.model)
            .where(self.model.id == id)
            .values(**updates)
        )
        await self.session.flush()
        return await self.get(id)
    
    async def delete(self, id: UUID) -> bool:
        """Delete entity."""
        result = await self.session.execute(
            delete(self.model).where(self.model.id == id)
        )
        return result.rowcount > 0

Step 2: Repository Specifico

# app/repositories/story_repository.py

from typing import List, Optional
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.orm import joinedload

from app.models import Story
from app.repositories.base_repository import BaseRepository

class StoryRepository(BaseRepository[Story]):
    """Repository per Story con query specifiche."""
    
    def __init__(self, session):
        super().__init__(session, Story)
    
    async def get_user_stories(
        self,
        user_id: UUID,
        limit: int = 20
    ) -> List[Story]:
        """Get all stories for a user."""
        result = await self.session.execute(
            select(Story)
            .where(Story.user_id == user_id)
            .order_by(Story.created_at.desc())
            .limit(limit)
        )
        return list(result.scalars().all())
    
    async def get_with_user(self, story_id: UUID) -> Optional[Story]:
        """Get story with user data (eager loading)."""
        result = await self.session.execute(
            select(Story)
            .options(joinedload(Story.user))
            .where(Story.id == story_id)
        )
        return result.unique().scalar_one_or_none()

Step 3: Service Layer

# app/services/story_service.py

from typing import Optional
from uuid import UUID
from app.repositories.story_repository import StoryRepository
from app.repositories.user_repository import UserRepository
from app.schemas import StoryCreateRequest
from app.models import Story

class StoryService:
    """Service per business logic delle storie."""
    
    def __init__(
        self,
        story_repo: StoryRepository,
        user_repo: UserRepository
    ):
        self.story_repo = story_repo
        self.user_repo = user_repo
    
    async def create_story(
        self,
        request: StoryCreateRequest,
        user_id: UUID
    ) -> Story:
        """Create story con business logic."""
        
        # Validate user exists
        user = await self.user_repo.get(user_id)
        if not user:
            raise ValueError("User not found")

        # Business logic: emotional score calculation
        score = self._calculate_emotional_score(
            request.emotional_state,
            request.previous_story_id
        )

        # Create via repository
        story = Story(
            user_id=user_id,
            title=request.title,
            content=request.content,
            emotional_state=request.emotional_state,
            emotional_score=score
        )

        story = await self.story_repo.create(story)

        # Update user stats via repository
        await self.user_repo.update_stats(user_id, score)
        
        return story
    
    def _calculate_emotional_score(
        self,
        emotional_state: str,
        previous_story_id: Optional[UUID]
    ) -> float:
        """Emotional score calculation."""

        # Emotion intensity mapping
        intensity = {
            "joy": 8.0,
            "hope": 7.0,
            "calm": 6.0,
            "sadness": 5.0,
            "anxiety": 4.0
        }

        base_score = intensity.get(emotional_state, 5.0)

        # Progression bonus if previous story exists
        if previous_story_id:
            # Real implementation: fetch previous story and calculate progression
            base_score *= 1.2
        
        return base_score

Step 4: Dependency Injection

# app/dependencies.py

from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_db
from app.repositories.story_repository import StoryRepository
from app.repositories.user_repository import UserRepository
from app.services.story_service import StoryService

def get_story_repository(
    session: AsyncSession = Depends(get_db)
) -> StoryRepository:
    return StoryRepository(session)

def get_user_repository(
    session: AsyncSession = Depends(get_db)
) -> UserRepository:
    return UserRepository(session)

def get_story_service(
    story_repo: Annotated[StoryRepository, Depends(get_story_repository)],
    user_repo: Annotated[UserRepository, Depends(get_user_repository)]
) -> StoryService:
    return StoryService(story_repo, user_repo)

StoryServiceDep = Annotated[StoryService, Depends(get_story_service)]

Step 5: Router Pulito

# app/routers/stories.py

from fastapi import APIRouter, HTTPException
from uuid import UUID

from app.dependencies import StoryServiceDep
from app.schemas import StoryResponse, StoryCreateRequest

router = APIRouter(prefix="/stories", tags=["stories"])

@router.post("/", response_model=StoryResponse)
async def create_story(
    request: StoryCreateRequest,
    story_service: StoryServiceDep,
    current_user: User = Depends(get_current_user)
):
    """Create new story."""
    try:
        story = await story_service.create_story(request, current_user.id)
        return story
    except ValueError as e:
        raise HTTPException(400, str(e))

@router.get("/{story_id}", response_model=StoryResponse)
async def get_story(
    story_id: UUID,
    story_service: StoryServiceDep
):
    """Get story by ID."""
    story = await story_service.get_story(story_id)
    
    if not story:
        raise HTTPException(404, "Story not found")
    
    return story

Testing: Il Vero Vantaggio

Test con Mock (Veloce)

# test_story_service.py

from unittest.mock import Mock, AsyncMock
import pytest
from uuid import uuid4

from app.services.story_service import StoryService

@pytest.fixture
def mock_story_repo():
    return AsyncMock()

@pytest.fixture
def mock_user_repo():
    return AsyncMock()

@pytest.mark.asyncio
async def test_calculate_emotional_score():
    """Test business logic pura - zero database."""
    service = StoryService(Mock(), Mock())
    
    score = service._calculate_emotional_score("joy", None)
    assert score == 8.0
    
    score = service._calculate_emotional_score("joy", uuid4())
    assert score == 9.6  # 8.0 * 1.2

@pytest.mark.asyncio
async def test_create_story(mock_story_repo, mock_user_repo):
    """Test orchestrazione con repository mockati."""
    
    user = User(id=uuid4())
    mock_user_repo.get.return_value = user
    
    story = Story(id=uuid4(), emotional_score=8.0)
    mock_story_repo.create.return_value = story
    
    service = StoryService(mock_story_repo, mock_user_repo)
    
    request = StoryCreateRequest(
        title="Test",
        content="Content",
        emotional_state="joy"
    )
    
    result = await service.create_story(request, user.id)
    
    assert result.emotional_score == 8.0
    mock_story_repo.create.assert_called_once()
    mock_user_repo.update_stats.assert_called_once()

I test sono veloci perché non richiedono database.

Errori Comuni

Errore 1: Business Logic nel Repository

# Wrong
class StoryRepository:
    async def create_with_score(self, story):
        # NO! Business logic non va qui
        score = calculate_score(story.emotional_state)
        story.emotional_score = score
        return await self.create(story)

# Correct
class StoryService:
    async def create_story(self, data):
        # Business logic in the service
        score = self._calculate_score(data.emotional_state)
        
        story = Story(emotional_score=score, **data.model_dump())
        return await self.story_repo.create(story)

Regola: Repository = solo SQL. Service = logica.

Errore 2: Repository che Chiama Repository

# Wrong
class StoryRepository:
    async def create_and_update_user(self, story):
        # NO! Repositories must not orchestrate
        user_repo = UserRepository(self.session)
        await user_repo.update(story.user_id)
        return await self.create(story)

# Correct
class StoryService:
    async def create_story(self, data):
        # Service coordinates repositories
        story = await self.story_repo.create(data)
        await self.user_repo.update(data.user_id)
        return story

Regola: Repository non parlano tra loro. Service li coordina.

Errore 3: Dimenticare None Check

# Risk of crash
@router.get("/stories/{story_id}")
async def get_story(story_id: UUID, service: StoryService):
    story = await service.get_story(story_id)
    return story  # If None, returns 500 instead of 404

# Correct
@router.get("/stories/{story_id}")
async def get_story(story_id: UUID, service: StoryService):
    story = await service.get_story(story_id)
    
    if not story:
        raise HTTPException(404, "Story not found")
    
    return story

Quando Usare Repository Pattern

Usalo Se

Business logic complessa: Se calcoli, validazioni, orchestrazioni multiple sono presenti

Test importanti: Se vuoi testare business logic senza setup database

Team in crescita: Se più developer lavorano sul progetto

Previsione cambio database: Se pensi di migrare (es. SQLite → PostgreSQL)

Query ripetute: Se stessa query usata in più endpoint

Non Usarlo Se

Prototipo rapido: MVP da buttare dopo, demo, proof of concept

CRUD semplicissimo: Solo get/create/update/delete senza logica

Progetto personale piccolo: Se hai 2-3 modelli e resterà così

Script one-off: Task amministrativi, migrazioni, batch jobs

Segnali che È Ora di Introdurlo

• 🚨 Stessa query copiata in 3+ posti
• 🚨 Router con >50 righe
• 🚨 Business logic mischiata a SQL
• 🚨 Paura di modificare query

Regola pratica: Se questi problemi fanno male, usa Repository. Altrimenti, YAGNI (You Aren’t Gonna Need It).

Conclusione

Repository Pattern risolve problemi reali:

Separazione responsabilità:

• Router = HTTP
• Service = business logic
• Repository = SQL

Testing facilitato: Business logic testabile senza database

Manutenibilità: Cambio database? Tocchi solo repository.

Quando usarlo: Quando business logic è complessa e test sono importanti.

Quando NON usarlo: Quando semplicità e velocità contano più di architettura.

Prossimi Passi

Per iniziare:

1. Identifica endpoint più complesso
2. Estrai business logic in Service
3. Crea Repository per query
4. Scrivi test con mock

Codice completo: GitHub: fastapi-repository-pattern

Include:

• ✅ BaseRepository completo
• ✅ Esempi repository specifici
• ✅ Service con business logic
• ✅ Test con mock
• ✅ Docker setup

---

📅 Pubblicato: 15 Gennaio 2025

⏱️ Tempo lettura: 12 minuti

🏷️ Tag: #FastAPI #Python #Architecture #Testing

---

Il Repository Pattern non è magia. È separazione delle responsabilità applicata al database. Usalo quando serve, ignoralo quando non serve.

Build pragmatically.