Use Cases Pattern

Cos'è un Use Case?

Un Use Case (o Interactor) rappresenta una singola operazione di business. È il punto di ingresso per tutta la logica applicativa.

Struttura Standard

# application/use_cases/create_device.py
from dataclasses import dataclass
from typing import Optional
from domain.entities.device import Device
from domain.ports.outbound.device_repository import DeviceRepository

# 1. Request DTO
@dataclass
class CreateDeviceRequest:
    user_id: int
    name: str
    protocol: str
    unique_id: str

# 2. Response DTO
@dataclass
class CreateDeviceResponse:
    device: Device

# 3. Use Case Class
class CreateDeviceUseCase:
    """
    Crea un nuovo device per l'utente.
    
    Responsabilità:
    - Validare business rules
    - Orchestrare le operazioni
    - Gestire le transazioni
    """
    
    def __init__(
        self,
        device_repository: DeviceRepository,
        billing_client: BillingClient = None,
    ):
        self._device_repo = device_repository
        self._billing = billing_client
    
    async def execute(self, request: CreateDeviceRequest) -> CreateDeviceResponse:
        # Business validation
        if not request.unique_id:
            raise ValueError("Device unique_id is required")
        
        # Check existing
        existing = await self._device_repo.find_by_unique_id(request.unique_id)
        if existing:
            raise DeviceAlreadyExistsError(request.unique_id)
        
        # Check license (cross-service call)
        if self._billing:
            has_license = await self._billing.check_license(request.user_id)
            if not has_license:
                raise NoLicenseAvailableError()
        
        # Create entity
        device = Device(
            id=None,
            name=request.name,
            protocol=request.protocol,
            unique_id=request.unique_id,
            owner_id=request.user_id,
        )
        
        # Persist
        saved = await self._device_repo.save(device)
        
        return CreateDeviceResponse(device=saved)

Vantaggi

1. Single Responsibility

Ogni Use Case fa una sola cosa:

• CreateDeviceUseCase - crea device
• GetDeviceUseCase - recupera device
• SuspendDeviceUseCase - sospende device

2. Testabilità Isolata

# tests/unit/test_create_device.py
import pytest
from unittest.mock import AsyncMock

async def test_create_device_success():
    # Arrange
    mock_repo = AsyncMock()
    mock_repo.find_by_unique_id.return_value = None
    mock_repo.save.return_value = Device(id=DeviceId(1), ...)
    
    use_case = CreateDeviceUseCase(device_repository=mock_repo)
    
    # Act
    response = await use_case.execute(CreateDeviceRequest(
        user_id=1,
        name="Test Device",
        protocol="huabao",
        unique_id="ABC123"
    ))
    
    # Assert
    assert response.device.id.value == 1
    mock_repo.save.assert_called_once()

3. Riutilizzabilità

Lo stesso Use Case può essere invocato da:

• HTTP API (FastAPI route)
• CLI (typer command)
• Worker (Redis consumer)
• GraphQL (resolver)
• Scheduled Job (cron)

Pattern di Composizione

Multiple Repositories

class TransferDeviceUseCase:
    def __init__(
        self,
        device_repo: DeviceRepository,
        user_device_repo: UserDeviceRepository,
        notification_service: NotificationService,
    ):
        self._device_repo = device_repo
        self._user_device_repo = user_device_repo
        self._notifications = notification_service

Cross-Service Communication

class ReactivateDeviceUseCase:
    def __init__(
        self,
        device_repo: DeviceRepository,
        billing_client: BillingClient,  # Calls billing service
    ):
        ...
    
    async def execute(self, request):
        # Check license on billing service
        license_ok = await self._billing_client.consume_license(
            user_id=request.user_id
        )
        if not license_ok:
            raise NoLicenseError()
        
        # Proceed with reactivation
        ...

Gestione Errori

Definisci eccezioni di dominio:

# domain/exceptions.py
class DomainError(Exception):
    """Base class for domain errors."""
    pass

class DeviceNotFoundError(DomainError):
    pass

class DeviceAlreadyExistsError(DomainError):
    def __init__(self, unique_id: str):
        super().__init__(f"Device with unique_id {unique_id} already exists")

class AccessDeniedError(DomainError):
    pass

class NoLicenseAvailableError(DomainError):
    pass

Mappale a HTTP nei routes:

# routes/devices.py
@router.post("/")
async def create_device(data: CreateDeviceBody, user = Depends(get_current_user)):
    try:
        response = await use_case.execute(CreateDeviceRequest(...))
        return response.device.to_dict()
    except DeviceAlreadyExistsError as e:
        raise HTTPException(409, str(e))
    except NoLicenseAvailableError:
        raise HTTPException(402, "No license available")
    except AccessDeniedError:
        raise HTTPException(403, "Access denied")

Naming Convention

Operazione	Nome Use Case
Lettura singola	GetDeviceUseCase
Lettura lista	ListDevicesUseCase o GetUserDevicesUseCase
Creazione	CreateDeviceUseCase o ClaimDeviceUseCase
Aggiornamento	UpdateDeviceUseCase
Eliminazione	DeleteDeviceUseCase o UnclaimDeviceUseCase
Azione specifica	SuspendDeviceUseCase, ReactivateDeviceUseCase

Checklist

• Request e Response sono @dataclass
• Use Case riceve dipendenze nel __init__
• Nessun import di FastAPI, SQLAlchemy nel Use Case
• Business errors sono eccezioni di dominio
• Metodo principale è execute() o __call__()