Structum DI (`structum-di`)¶

Structum DI implementa un container IoC enterprise (basato su dependency-injector) per la gestione delle dipendenze.

Feature	Stato	Versione
Stato	Alpha	0.1.0
Namespace	`structum_lab.plugins.di`
Core	`dependency-injector`

Indice¶

Cos’è DI Plugin
Concetti Fondamentali
Quick Start (5 Minuti)
StructumContainer
Provider Types
FastAPI Integration
Advanced Patterns
Testing
Best Practices
API Reference

1. Cos’è DI Plugin¶

structum-di fornisce un container Dependency Injection enterprise-grade basato su dependency-injector.

1.1 Problema Risolto¶

Prima (senza DI):

# ❌ Hard-coded dependencies
class UserService:
    def __init__(self):
        self.db = PostgresDatabase("postgresql://...")  # Tight coupling
        self.logger = Logger("app")  # Hard to test
        self.config = {...}  # No validation

# ❌ Difficile testare
# ❌ Difficile cambiare implementazioni
# ❌ Boilerplate per passare dipendenze

Dopo (con DI):

# ✅ Dependency injection
class UserService:
    def __init__(self, db: DatabaseInterface, logger: LoggerInterface, config: Config):
        self.db = db
        self.logger = logger
        self.config = config

# Container risolve automaticamente
container = AppContainer()
service = container.user_service()  # Tutte le dipendenze iniettate!

1.2 Caratteristiche Principali¶

Feature	Descrizione
Pure IoC	No dipendenze hard-coded - tutto configurabile
Type Safe	Full type hints per mypy strict
Modular	Installa solo le dipendenze necessarie
Testable	Easy mocking e dependency override
FastAPI Ready	Integration patterns per web apps
Lifecycle Management	Singleton, Factory, Transient scopes

2. Concetti Fondamentali¶

2.1 Inversion of Control (IoC)¶

┌─────────────────────────────────────────┐
│ Traditional (Direct Dependencies)       │
├─────────────────────────────────────────┤
│ UserService                             │
│   ├─ creates Database()                │
│   ├─ creates Logger()                  │
│   └─ creates Config()                  │
│                                         │
│ Problem: Tight coupling!                │
└─────────────────────────────────────────┘

┌─────────────────────────────────────────┐
│ IoC with DI (Injected Dependencies)     │
├─────────────────────────────────────────┤
│ Container                               │
│   ├─ provides Database                 │
│   ├─ provides Logger                   │
│   └─ provides Config                   │
│         ↓                               │
│   UserService(db, logger, config)      │
│                                         │
│ Benefit: Loose coupling!                │
└─────────────────────────────────────────┘

2.2 Provider Scopes¶

Scope	Behavior	Use Case
Singleton	Single instance per container	Database, Config, Logger
Factory	New instance per call	Request handlers, Services
Transient	New instance always	Disposable objects

2.3 Wiring¶

Wiring = automatic dependency injection in functions/classes:

from dependency_injector.wiring import inject, Provide

@inject
def create_user(
    username: str,
    service: UserService = Depends(Provide[Container.user_service])
):
    return service.create(username)

3. Quick Start (5 Minuti)¶

Step 1: Installazione¶

pip install -e packages/di

Step 2: Definisci Container¶

from dependency_injector import containers, providers
from structum_lab.plugins.di import StructumContainer
from structum_lab.config import get_config
from structum_lab.logging import get_logger

class AppContainer(StructumContainer):
    """Application dependency container."""
    
    # Core dependencies (override abstract base)
    config = providers.Singleton(get_config)
    logger = providers.Factory(get_logger)
    
    # Application services
    user_service = providers.Factory(
        UserService,
        db=providers.Dependency(),  # Will be injected
        logger=logger("user_service"),
        config=config
    )
    
    payment_service = providers.Factory(
        PaymentService,
        api_key=config.provided.payment.api_key,
        logger=logger("payment")
    )

Step 3: Use Container¶

# Initialize container
container = AppContainer()

# Wire modules for @inject decorator
container.wire(modules=[__name__])

# Get service instance
user_service = container.user_service()

# Use service
user = user_service.create_user("john_doe")

4. StructumContainer¶

4.1 Base Container¶

StructumContainer fornisce dependencies astratte per Structum core:

from structum_lab.plugins.di import StructumContainer

class StructumContainer(containers.DeclarativeContainer):
    """Base container con core dependencies."""
    
    # Abstract providers (must override)
    config = providers.Dependency()
    logger = providers.Dependency()
    
    # Optional providers
    database = providers.Dependency()
    metrics = providers.Dependency()

4.2 Overriding Core Dependencies¶

from structum_lab.plugins.dynaconf import DynaconfConfigProvider
from structum_lab.plugins.observability import StructuredLogger

class AppContainer(StructumContainer):
    """Override abstract dependencies."""
    
    # Concrete implementations
    config = providers.Singleton(
        DynaconfConfigProvider,
        root_path=".",
        env_prefix="MYAPP"
    )
    
    logger = providers.Factory(
        StructuredLogger,
        name="myapp"
    )

4.3 Modular Containers¶

class DatabaseContainer(containers.DeclarativeContainer):
    """Database services container."""
    
    config = providers.Dependency()
    logger = providers.Dependency()
    
    database = providers.Singleton(
        SQLAlchemyDatabase.from_config
    )
    
    user_repo = providers.Factory(
        UserRepository,
        db=database
    )
    
    post_repo = providers.Factory(
        PostRepository,
        db=database
    )

class AppContainer(StructumContainer):
    """Main container composing other containers."""
    
    config = providers.Singleton(get_config)
    logger = providers.Factory(get_logger)
    
    # Compose database container
    database_container = providers.Container(
        DatabaseContainer,
        config=config,
        logger=logger
    )
    
    # Access nested providers
    user_service = providers.Factory(
        UserService,
        user_repo=database_container.user_repo
    )

5. Provider Types¶

5.1 Singleton Provider¶

Single instance shared across application:

class AppContainer(containers.DeclarativeContainer):
    # Database connection (expensive, reuse)
    database = providers.Singleton(
        SQLAlchemyDatabase,
        url="postgresql://..."
    )
    
    # Config (immutable, reuse)
    config = providers.Singleton(get_config)

5.2 Factory Provider¶

New instance per call:

class AppContainer(containers.DeclarativeContainer):
    # Logger factory (different name per component)
    logger = providers.Factory(
        get_logger
    )
    
    # Service (new instance per request)
    user_service = providers.Factory(
        UserService,
        db=database,
        logger=logger("users")  # Factory called with argument
    )

5.3 Configuration Provider¶

Access config values:

from dependency_injector import providers

class AppContainer(containers.DeclarativeContainer):
    config = providers.Configuration()
    
    # Load from dict
    config.from_dict({
        "database": {
            "url": "postgresql://...",
            "pool_size": 10
        }
    })
    
    # Use in other providers
    database = providers.Singleton(
        SQLAlchemyDatabase,
        url=config.database.url,
        pool_size=config.database.pool_size
    )

5.4 Callable Provider¶

Wrap callables:

class AppContainer(containers.DeclarativeContainer):
    # Wrap function
    hash_password = providers.Callable(
        hashlib.sha256
    )
    
    # Use in services
    auth_service = providers.Factory(
        AuthService,
        hasher=hash_password
    )

5.5 Resource Provider¶

Managed resources with initialization/shutdown:

def init_database(url: str):
    """Initialize database resource."""
    db = SQLAlchemyDatabase(url)
    yield db
    db.shutdown()  # Cleanup

class AppContainer(containers.DeclarativeContainer):
    config = providers.Configuration()
    
    database = providers.Resource(
        init_database,
        url=config.database.url
    )

6. FastAPI Integration¶

6.1 Application Setup¶

from fastapi import FastAPI
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide

# Container definition
class Container(StructumContainer):
    config = providers.Singleton(get_config)
    logger = providers.Factory(get_logger)
    
    database = providers.Singleton(
        SQLAlchemyDatabase.from_config
    )
    
    user_service = providers.Factory(
        UserService,
        db=database,
        logger=logger("users")
    )

# FastAPI app
app = FastAPI()

# Initialize container
container = Container()
container.wire(modules=[__name__])

@app.on_event("startup")
async def startup():
    """Initialize resources."""
    await container.init_resources()

@app.on_event("shutdown")
async def shutdown():
    """Cleanup resources."""
    await container.shutdown_resources()

6.2 Dependency Injection in Routes¶

from fastapi import Depends

@app.get("/users/{user_id}")
@inject
async def get_user(
    user_id: int,
    service: UserService = Depends(Provide[Container.user_service])
):
    """Get user by ID."""
    user = service.get_by_id(user_id)
    
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    return {"id": user.id, "username": user.username}

@app.post("/users")
@inject
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(Provide[Container.user_service]),
    logger: LoggerInterface = Depends(Provide[Container.logger])
):
    """Create new user."""
    logger.info("Creating user", username=user_data.username)
    
    user = service.create(user_data.username, user_data.email)
    
    logger.info("User created", user_id=user.id)
    
    return {"id": user.id}

6.3 Request-Scoped Dependencies¶

from dependency_injector import providers

class Container(StructumContainer):
    # ... other providers ...
    
    # Request-scoped logger with correlation ID
    request_logger = providers.Factory(
        lambda request_id: get_logger("api").bind(request_id=request_id)
    )

@app.middleware("http")
async def correlation_id_middleware(request: Request, call_next):
    """Add correlation ID to requests."""
    import uuid
    request.state.correlation_id = str(uuid.uuid4())
    
    response = await call_next(request)
    response.headers["X-Correlation-ID"] = request.state.correlation_id
    
    return response

@app.post("/users")
@inject
async def create_user(
    request: Request,
    user_data: UserCreate,
    service: UserService = Depends(Provide[Container.user_service])
):
    # Get request-scoped logger
    logger = container.request_logger(request.state.correlation_id)
    logger.info("Creating user", username=user_data.username)
    
    user = service.create(user_data.username, user_data.email)
    
    return {"id": user.id}

7. Advanced Patterns¶

7.1 Lazy Initialization¶

from dependency_injector import providers

class Container(containers.DeclarativeContainer):
    # Expensive resource (load only when needed)
    ml_model = providers.Singleton(
        load_ml_model,
        model_path="/models/large_model.pkl"
    )
    
    # Service that may or may not need model
    prediction_service = providers.Factory(
        PredictionService,
        model=ml_model.provider  # Lazy reference
    )

7.2 Conditional Providers¶

import os

class Container(containers.DeclarativeContainer):
    # Different implementations per environment
    if os.getenv("ENV") == "production":
        cache = providers.Singleton(
            RedisCache,
            url="redis://prod.cache.internal"
        )
    else:
        cache = providers.Singleton(
            InMemoryCache
        )

7.3 Factory with Arguments¶

class Container(containers.DeclarativeContainer):
    logger = providers.Factory(get_logger)
    
    # Factory that accepts runtime arguments
    service_factory = providers.FactoryAggregate(
        user_service=providers.Factory(
            UserService,
            logger=logger("users")
        ),
        payment_service=providers.Factory(
            PaymentService,
            logger=logger("payments")
        )
    )

# Usage
user_service = container.service_factory("user_service")
payment_service = container.service_factory("payment_service")

7.4 Override for Testing¶

# Production container
class ProductionContainer(StructumContainer):
    database = providers.Singleton(
        SQLAlchemyDatabase,
        url="postgresql://prod.db.internal/app"
    )

# Test override
def test_user_creation():
    container = ProductionContainer()
    
    # Override with mock
    mock_db = MockDatabase()
    container.database.override(mock_db)
    
    # Test with mock
    service = container.user_service()
    user = service.create("test_user")
    
    assert user.username == "test_user"
    assert len(mock_db.queries) == 1

8. Testing¶

8.1 Test Container¶

import pytest
from unittest.mock import Mock

class TestContainer(containers.DeclarativeContainer):
    """Test-specific container with mocks."""
    
    config = providers.Singleton(
        lambda: {"database": {"url": "sqlite:///:memory:"}}
    )
    
    logger = providers.Factory(
        lambda name: Mock(spec=LoggerInterface)
    )
    
    database = providers.Singleton(
        lambda: Mock(spec=DatabaseInterface)
    )
    
    user_service = providers.Factory(
        UserService,
        db=database,
        logger=logger("users"),
        config=config
    )

@pytest.fixture
def container():
    """Provide test container."""
    c = TestContainer()
    yield c
    c.reset_singletons()

def test_user_service(container):
    """Test user service with mocked dependencies."""
    service = container.user_service()
    
    # Mock database response
    container.database().query.return_value = [
        {"id": 1, "username": "test_user"}
    ]
    
    users = service.get_all()
    
    assert len(users) == 1
    assert users[0].username == "test_user"

8.2 Override Dependencies¶

def test_with_override():
    """Test with dependency override."""
    container = AppContainer()
    
    # Create mock
    mock_logger = Mock()
    
    # Override logger provider
    with container.logger.override(lambda name: mock_logger):
        service = container.user_service()
        service.create_user("test")
        
        # Verify mock was called
        mock_logger.info.assert_called()

8.3 Integration Test Container¶

class IntegrationTestContainer(StructumContainer):
    """Container for integration tests."""
    
    config = providers.Singleton(get_config)
    logger = providers.Factory(get_logger)
    
    # Real database for integration tests
    database = providers.Singleton(
        SQLAlchemyDatabase,
        url="postgresql://localhost/test_db"
    )
    
    user_service = providers.Factory(
        UserService,
        db=database,
        logger=logger("users")
    )

@pytest.fixture(scope="session")
def integration_container():
    """Setup integration test environment."""
    container = IntegrationTestContainer()
    
    # Run migrations
    db = container.database()
    run_migrations(db)
    
    yield container
    
    # Cleanup
    db.shutdown()

def test_user_creation_integration(integration_container):
    """Integration test with real database."""
    service = integration_container.user_service()
    
    user = service.create_user("integration_test_user")
    
    # Verify in database
    db = integration_container.database()
    with db.transaction() as conn:
        result = conn.execute(
            "SELECT * FROM users WHERE id = :id",
            {"id": user.id}
        )
        db_user = result.fetchone()
        
        assert db_user["username"] == "integration_test_user"

9. Best Practices¶

9.1 Container Organization¶

# ❌ Bad: Everything in one container
class GiantContainer(containers.DeclarativeContainer):
    # 100+ providers...
    pass

# ✅ Good: Modular containers
class DatabaseContainer(containers.DeclarativeContainer):
    """Database-related dependencies."""
    pass

class CacheContainer(containers.DeclarativeContainer):
    """Cache-related dependencies."""
    pass

class AppContainer(StructumContainer):
    """Main container composing modules."""
    database = providers.Container(DatabaseContainer)
    cache = providers.Container(CacheContainer)

9.2 Avoid Service Locator Pattern¶

# ❌ Bad: Service Locator (anti-pattern)
class UserService:
    def create_user(self, username):
        db = container.database()  # Looking up in function
        logger = container.logger()  # Hard to test
        
        # ...

# ✅ Good: Constructor Injection
class UserService:
    def __init__(self, db: DatabaseInterface, logger: LoggerInterface):
        self.db = db
        self.logger = logger
    
    def create_user(self, username):
        # Dependencies already available
        self.logger.info("Creating user", username=username)
        # ...

9.3 Interface Segregation¶

# ✅ Define interfaces for dependencies
from typing import Protocol

class EmailSenderInterface(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

class UserService:
    def __init__(
        self,
        db: DatabaseInterface,
        logger: LoggerInterface,
        email: EmailSenderInterface  # Interface, not concrete class
    ):
        self.db = db
        self.logger = logger
        self.email = email

10. API Reference¶

10.1 StructumContainer¶

class StructumContainer(containers.DeclarativeContainer):
    """Base container for Structum applications."""
    
    # Abstract providers (override in subclass)
    config: providers.Dependency
    logger: providers.Dependency
    
    # Optional providers
    database: providers.Dependency | None
    metrics: providers.Dependency | None

10.2 Common Provider Types¶

# Singleton: single instance
providers.Singleton(Class, *args, **kwargs)

# Factory: new instance per call
providers.Factory(Class, *args, **kwargs)

# Callable: wrap function
providers.Callable(function, *args, **kwargs)

# Resource: managed lifecycle
providers.Resource(init_function, *args, **kwargs)

# Configuration: config values
providers.Configuration()

# Dependency: abstract dependency
providers.Dependency()

# Container: nested container
providers.Container(ContainerClass, *args, **kwargs)

10.3 Wiring Decorator¶

from dependency_injector.wiring import inject, Provide

@inject
def function(
    arg1: str,
    service: ServiceClass = Depends(Provide[Container.service])
):
    """Function with injected dependency."""
    ...

Structum DI (structum-di)¶