Clean Architecture per Microservizi FastAPI

Introduzione

Questa guida documenta l'architettura Clean/Hexagonal adottata in tutti i microservizi FastAPI del progetto VISLA. L'obiettivo è garantire:

Separazione delle responsabilità - Ogni layer ha un compito specifico
Testabilità - Business logic isolata e facilmente mockabile
Manutenibilità - Cambiamenti in un layer non impattano gli altri
Indipendenza dal framework - Il domain non conosce FastAPI, SQLAlchemy, ecc.

Struttura Ideale di un Microservizio

service-name/
├── app.py                          # Entry point FastAPI
├── config.py                       # ⚠️ LEGACY - spostare in infrastructure/config/
│
├── domain/                         # 🟢 CORE - Business Logic Pura
│   ├── entities/                   # Modelli di dominio (dataclass)
│   │   ├── __init__.py
│   │   └── device.py               # @dataclass Device, DeviceId
│   │
│   └── ports/                      # Interfacce/Contratti
│       ├── inbound/                # Use Case interfaces (opzionale)
│       │   └── device_service.py   # Protocol/ABC per use cases
│       └── outbound/               # Repository interfaces
│           ├── __init__.py
│           └── device_repository.py # ABC DeviceRepository
│
├── application/                    # 🔵 USE CASES - Orchestrazione
│   ├── __init__.py
│   └── use_cases/
│       ├── __init__.py
│       ├── get_device.py           # GetDeviceUseCase
│       ├── create_device.py        # CreateDeviceUseCase
│       └── update_device.py        # UpdateDeviceUseCase
│
├── infrastructure/                 # 🟠 ADAPTERS - Implementazioni Concrete
│   ├── config/
│   │   ├── __init__.py
│   │   ├── settings.py             # Pydantic Settings (→ SPOSTARE config.py qui)
│   │   ├── database.py             # SessionLocal, engine, get_db
│   │   └── container.py            # DI Container
│   │
│   └── adapters/
│       ├── inbound/                # Driving adapters (API, CLI, Workers)
│       │   └── redis_consumer.py   # Background workers
│       │
│       ├── outbound/               # Driven adapters (DB, External APIs)
│       │   ├── orm_models.py       # SQLAlchemy models
│       │   ├── postgres_device_repository.py
│       │   └── redis_cache.py
│       │
│       └── clients/                # External service clients
│           └── billing_client.py
│
├── routes/                         # 🟣 HTTP ADAPTERS - FastAPI Routes
│   ├── __init__.py
│   ├── devices.py                  # APIRouter endpoints
│   ├── health.py
│   └── schema.py                   # Route per schema/migrations
│
├── tests/                          # Test Suite
│   ├── unit/                       # Unit tests (mocked)
│   ├── integration/                # Integration tests (DB)
│   └── conftest.py
│
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── pytest.ini
└── README.md

Cosa Rimuovere/Spostare

Nel tuo screenshot vedo file che andrebbero riorganizzati:

File Attuale	Posizione Ideale	Azione
`config.py` (root)	`infrastructure/config/settings.py`	Spostare
`routes/` (root)	OK ✅	Mantenere
`app.py` (root)	OK ✅	Mantenere (entry point)

Nota: app.py e Dockerfile rimangono nella root perché sono entry point. Ma la configurazione deve stare in infrastructure/ perché è un dettaglio implementativo.

I Layer Spiegati

🟢 Domain Layer (Centro)

Il cuore dell'applicazione. Contiene:

Entities: Oggetti di business puri (no dipendenze esterne)
Ports: Interfacce/contratti che definiscono COSA fare, non COME

# domain/entities/device.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class DeviceId:
    value: int

@dataclass
class Device:
    id: Optional[DeviceId]
    name: str
    protocol: str
    unique_id: str
    
    def to_dict(self) -> dict:
        return {
            "id": self.id.value if self.id else None,
            "name": self.name,
            "protocol": self.protocol,
        }

# domain/ports/outbound/device_repository.py
from abc import ABC, abstractmethod
from domain.entities.device import Device, DeviceId

class DeviceRepository(ABC):
    @abstractmethod
    async def find_by_id(self, device_id: DeviceId) -> Optional[Device]:
        pass
    
    @abstractmethod
    async def save(self, device: Device) -> Device:
        pass

Regola d'Oro: Il Domain Layer NON importa mai nulla da infrastructure/, routes/, SQLAlchemy, FastAPI, ecc.

🔵 Application Layer (Use Cases)

Contiene la logica di orchestrazione. Ogni Use Case:

Riceve una Request (dataclass)
Restituisce una Response (dataclass)
Usa i Ports per interagire con l'esterno

# application/use_cases/get_device.py
from dataclasses import dataclass
from domain.entities.device import Device, DeviceId
from domain.ports.outbound.device_repository import DeviceRepository

@dataclass
class GetDeviceRequest:
    device_id: int
    user_id: int

@dataclass
class GetDeviceResponse:
    device: Device

class GetDeviceUseCase:
    def __init__(self, device_repository: DeviceRepository):
        self._repo = device_repository
    
    async def execute(self, request: GetDeviceRequest) -> GetDeviceResponse:
        device = await self._repo.find_by_id(DeviceId(request.device_id))
        if not device:
            raise DeviceNotFoundError()
        return GetDeviceResponse(device=device)

Vantaggi:

Testabile con mock repository
Non conosce HTTP, DB, Redis
Riutilizzabile in CLI, workers, ecc.

🟠 Infrastructure Layer (Adapters)

Implementazioni concrete dei Ports:

# infrastructure/adapters/outbound/postgres_device_repository.py
from domain.ports.outbound.device_repository import DeviceRepository
from domain.entities.device import Device, DeviceId
from infrastructure.adapters.orm_models import DeviceModel

class PostgresDeviceRepository(DeviceRepository):
    def __init__(self, session_factory):
        self._session_factory = session_factory
    
    async def find_by_id(self, device_id: DeviceId) -> Optional[Device]:
        with self._session_factory() as session:
            model = session.get(DeviceModel, device_id.value)
            if model:
                return self._to_entity(model)
            return None
    
    def _to_entity(self, model: DeviceModel) -> Device:
        return Device(
            id=DeviceId(model.id),
            name=model.name,
            protocol=model.protocol,
            unique_id=model.unique_id,
        )

Il DI Container assembla tutto:

# infrastructure/config/container.py
class Container:
    def __init__(self, session_factory):
        self._session_factory = session_factory
    
    @property
    def device_repository(self) -> DeviceRepository:
        return PostgresDeviceRepository(self._session_factory)
    
    @property
    def get_device_use_case(self) -> GetDeviceUseCase:
        return GetDeviceUseCase(device_repository=self.device_repository)

🟣 Routes (HTTP Adapters)

I routes FastAPI sono solo adattatori HTTP che:

Validano input (Pydantic)
Chiamano Use Cases
Formattano output

# routes/devices.py
from fastapi import APIRouter, Depends, HTTPException
from infrastructure.config.container import get_container
from application.use_cases.get_device import GetDeviceRequest

router = APIRouter()

@router.get("/{device_id}")
async def get_device(device_id: int, current_user = Depends(get_current_user)):
    container = get_container()
    use_case = container.get_device_use_case
    
    try:
        response = await use_case.execute(
            GetDeviceRequest(device_id=device_id, user_id=current_user.id)
        )
        return response.device.to_dict()
    except DeviceNotFoundError:
        raise HTTPException(404, "Device not found")

Dependency Rule

┌─────────────────────────────────────────────────┐
│                 ROUTES (HTTP)                   │
│    ↓ conosce ↓                                  │
├─────────────────────────────────────────────────┤
│              APPLICATION (Use Cases)            │
│    ↓ conosce ↓                                  │
├─────────────────────────────────────────────────┤
│                   DOMAIN                        │
│    entities, ports                              │
│    ⚠️ NON conosce nient'altro                   │
├─────────────────────────────────────────────────┤
│              INFRASTRUCTURE                     │
│   implementa i ports definiti in domain         │
└─────────────────────────────────────────────────┘

Le dipendenze puntano verso l'interno (verso il Domain). Mai al contrario.

Vantaggi Pratici

1. Testing Semplificato

# test_get_device.py
class MockDeviceRepository(DeviceRepository):
    async def find_by_id(self, device_id):
        return Device(id=DeviceId(1), name="Test", ...)

async def test_get_device_success():
    use_case = GetDeviceUseCase(device_repository=MockDeviceRepository())
    response = await use_case.execute(GetDeviceRequest(device_id=1, user_id=1))
    assert response.device.name == "Test"

2. Cambio Database Trasparente

Se domani vuoi passare da PostgreSQL a MongoDB, cambi solo l'adapter:

class MongoDeviceRepository(DeviceRepository):
    # Nuova implementazione
    pass

Il Use Case non cambia!

3. Riutilizzo Business Logic

Lo stesso Use Case può essere usato da:

HTTP API (FastAPI routes)
CLI commands
Background workers
GraphQL resolvers

Checklist Refactoring

config.py spostato in infrastructure/config/settings.py
Tutte le Entities sono @dataclass in domain/entities/
Repository definiti come ABC in domain/ports/outbound/
Ogni operazione ha un Use Case in application/use_cases/
ORM models solo in infrastructure/adapters/orm_models.py
Routes chiamano Use Cases, non repository direttamente
DI Container in infrastructure/config/container.py

Introduzione​

Struttura Ideale di un Microservizio​

Cosa Rimuovere/Spostare​

I Layer Spiegati​

🟢 Domain Layer (Centro)​

🔵 Application Layer (Use Cases)​

🟠 Infrastructure Layer (Adapters)​

🟣 Routes (HTTP Adapters)​

Dependency Rule​

Vantaggi Pratici​

1. Testing Semplificato​

2. Cambio Database Trasparente​

3. Riutilizzo Business Logic​

Checklist Refactoring​