Dependency Injection Container

Cos'è il DI Container?

Il Dependency Injection Container è responsabile di:

• Creare istanze di repository e use cases
• Gestire le dipendenze tra componenti
• Fornire un punto centrale per la configurazione

Implementazione Standard

# infrastructure/config/container.py
from typing import Callable
from sqlalchemy.orm import Session

from domain.ports.outbound.device_repository import DeviceRepository
from domain.ports.outbound.user_device_repository import UserDeviceRepository
from infrastructure.adapters.outbound.postgres_device_repository import PostgresDeviceRepository
from infrastructure.adapters.outbound.postgres_user_device_repository import PostgresUserDeviceRepository
from application.use_cases.get_device import GetDeviceUseCase
from application.use_cases.create_device import CreateDeviceUseCase
from application.use_cases.update_device import UpdateDeviceUseCase


class Container:
    """Dependency Injection container for the service."""

    def __init__(self, session_factory: Callable[[], Session]):
        self._session_factory = session_factory
        self._instances = {}  # Cache for singletons

    # ================== REPOSITORIES ==================
    
    @property
    def device_repository(self) -> DeviceRepository:
        """Lazy-loaded singleton repository."""
        if 'device_repository' not in self._instances:
            self._instances['device_repository'] = PostgresDeviceRepository(
                self._session_factory
            )
        return self._instances['device_repository']

    @property
    def user_device_repository(self) -> UserDeviceRepository:
        if 'user_device_repository' not in self._instances:
            self._instances['user_device_repository'] = PostgresUserDeviceRepository(
                self._session_factory
            )
        return self._instances['user_device_repository']

    # ================== USE CASES ==================
    
    @property
    def get_device_use_case(self) -> GetDeviceUseCase:
        """Creates a new use case instance with dependencies."""
        return GetDeviceUseCase(
            device_repository=self.device_repository,
            user_device_repository=self.user_device_repository,
        )

    @property
    def create_device_use_case(self) -> CreateDeviceUseCase:
        return CreateDeviceUseCase(
            device_repository=self.device_repository,
        )

    @property
    def update_device_use_case(self) -> UpdateDeviceUseCase:
        return UpdateDeviceUseCase(
            device_repository=self.device_repository,
            user_device_repository=self.user_device_repository,
        )


# ================== GLOBAL INSTANCE ==================

_container: Container = None


def init_container(session_factory: Callable[[], Session]) -> Container:
    """Initialize the global container. Call once at startup."""
    global _container
    _container = Container(session_factory)
    return _container


def get_container() -> Container:
    """Get the global container instance."""
    if _container is None:
        raise RuntimeError("Container not initialized. Call init_container() first.")
    return _container

Inizializzazione in app.py

# app.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

from infrastructure.config.database import SessionLocal
from infrastructure.config.container import init_container


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    init_container(SessionLocal)
    yield
    # Shutdown (cleanup if needed)


app = FastAPI(lifespan=lifespan)

Utilizzo nelle Routes

# routes/devices.py
from fastapi import APIRouter, Depends
from infrastructure.config.container import get_container

router = APIRouter()

@router.get("/{device_id}")
async def get_device(device_id: int, user = Depends(get_current_user)):
    container = get_container()
    use_case = container.get_device_use_case
    
    response = await use_case.execute(
        GetDeviceRequest(device_id=device_id, user_id=user.id)
    )
    return response.device.to_dict()

Pattern Avanzati

External Clients

Per servizi esterni, aggiungi client al container:

class Container:
    # ...
    
    @property
    def billing_client(self) -> BillingClient:
        if 'billing_client' not in self._instances:
            from shared.infrastructure.clients.billing_client import get_billing_client
            self._instances['billing_client'] = get_billing_client()
        return self._instances['billing_client']
    
    @property
    def auth_client(self) -> AuthClient:
        if 'auth_client' not in self._instances:
            from shared.infrastructure.auth import get_auth_client
            self._instances['auth_client'] = get_auth_client()
        return self._instances['auth_client']
    
    # Use case con dipendenze esterne
    @property
    def reactivate_device_use_case(self) -> ReactivateDeviceUseCase:
        return ReactivateDeviceUseCase(
            device_repository=self.device_repository,
            billing_client=self.billing_client,  # Cross-service
        )

Scoped vs Singleton

Tipo	Quando Usare	Esempio
Singleton	Repository, Clients	device_repository, billing_client
Transient	Use Cases	get_device_use_case (nuovo ogni volta)
Scoped	Per-request (raro)	Session-bound objects

@property
def device_repository(self) -> DeviceRepository:
    # SINGLETON: cached in _instances
    if 'device_repository' not in self._instances:
        self._instances['device_repository'] = PostgresDeviceRepository(...)
    return self._instances['device_repository']

@property
def get_device_use_case(self) -> GetDeviceUseCase:
    # TRANSIENT: new instance every time
    return GetDeviceUseCase(device_repository=self.device_repository)

Testing con Mock Container

# tests/conftest.py
import pytest
from unittest.mock import AsyncMock

class MockContainer:
    def __init__(self):
        self._device_repo = AsyncMock()
    
    @property
    def device_repository(self):
        return self._device_repo
    
    @property
    def get_device_use_case(self):
        from application.use_cases.get_device import GetDeviceUseCase
        return GetDeviceUseCase(device_repository=self._device_repo)

@pytest.fixture
def mock_container():
    return MockContainer()

Best Practices

1. Una property per componente - Facile da navigare e testare
2. Lazy loading - Crea istanze solo quando servono
3. Singleton per I/O - Repository e client sono costosi da creare
4. Transient per Use Cases - Nessuno stato condiviso tra richieste
5. Init esplicito - init_container() nel lifespan, non import-time