Provider Modules

Provider modules use factory functions and the @provider decorator to create complex service instances with dependency injection support.

🎯 What are Providers?

Providers are factory functions that create service instances, automatically receiving their dependencies through injection.

from injectq import Module, provider, InjectQ

class ServiceModule(Module):
    @provider
    def create_database_pool(self) -> DatabasePool:
        """Factory for database connection pool"""
        return DatabasePool(
            host="localhost",
            port=5432,
            max_connections=20
        )

    @provider
    def create_user_service(self, user_repo: IUserRepository, email_svc: IEmailService) -> IUserService:
        """Factory for user service with dependencies"""
        return UserService(user_repo, email_svc)

# Usage
container = InjectQ()
container.install(ServiceModule())

# Services are created with dependencies injected
user_service = container.get(IUserService)  # Gets UserService with injected dependencies

🔧 Creating Provider Methods

Basic Provider

from injectq import Module, provider

class DatabaseModule(Module):
    @provider
    def database_connection(self) -> IDatabaseConnection:
        """Create database connection"""
        return PostgresConnection(
            host="localhost",
            database="myapp"
        )

Provider with Dependencies

class ServiceModule(Module):
    @provider
    def user_repository(self, db: IDatabaseConnection) -> IUserRepository:
        """Create user repository with database dependency"""
        return SqlUserRepository(db)

    @provider
    def user_service(self, user_repo: IUserRepository, email_svc: IEmailService) -> IUserService:
        """Create user service with its dependencies"""
        return UserService(user_repo, email_svc)

Provider with Configuration

class ConfigurableModule(Module):
    def __init__(self, config: AppConfig):
        self.config = config

    @provider
    def database_pool(self) -> IDatabasePool:
        """Create database pool with configuration"""
        return DatabasePool(
            host=self.config.database_host,
            port=self.config.database_port,
            max_connections=self.config.max_connections
        )

    @provider
    def cache_service(self) -> ICache:
        """Create cache service with configuration"""
        if self.config.use_redis:
            return RedisCache(self.config.redis_url)
        else:
            return InMemoryCache()

🎨 Provider Patterns

Complex Object Creation

class InfrastructureModule(Module):
    @provider
    def message_queue(self) -> IMessageQueue:
        """Create message queue with retry logic"""
        queue = RabbitMQConnection(
            host="rabbitmq-server",
            port=5672,
            credentials=self._load_credentials()
        )

        # Configure retry policy
        queue.retry_policy = ExponentialBackoffRetry(
            max_attempts=5,
            base_delay=1.0
        )

        return queue

    @provider
    def payment_processor(self, mq: IMessageQueue, db: IDatabase) -> IPaymentProcessor:
        """Create payment processor with dependencies"""
        processor = StripePaymentProcessor(
            api_key=os.getenv("STRIPE_API_KEY"),
            message_queue=mq,
            database=db
        )

        # Configure webhooks
        processor.webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET")

        return processor

    def _load_credentials(self) -> Credentials:
        """Load MQ credentials from secure storage"""
        return Credentials(
            username=os.getenv("MQ_USER"),
            password=os.getenv("MQ_PASS")
        )

Conditional Provider

class EnvironmentModule(Module):
    def __init__(self, environment: str):
        self.environment = environment

    @provider
    def email_service(self) -> IEmailService:
        """Create email service based on environment"""
        if self.environment == "production":
            return SmtpEmailService(
                host="smtp.gmail.com",
                port=587,
                credentials=self._load_smtp_credentials()
            )
        elif self.environment == "testing":
            return MockEmailService()
        else:
            return ConsoleEmailService()  # Development

    @provider
    def cache_service(self) -> ICache:
        """Create cache service based on environment"""
        if self.environment == "production":
            return RedisCache(host="redis-cluster")
        else:
            return InMemoryCache()

Resource Management Provider

class ResourceModule(Module):
    @provider
    def database_connection_pool(self) -> IDatabasePool:
        """Create managed database connection pool"""
        pool = DatabasePool(
            host="localhost",
            max_connections=20,
            min_connections=5
        )

        # Register cleanup
        import atexit
        atexit.register(pool.close_all)

        return pool

    @provider
    def file_manager(self) -> IFileManager:
        """Create file manager with temp directory"""
        temp_dir = tempfile.mkdtemp(prefix="app_")

        manager = FileManager(temp_dir)

        # Register cleanup
        import atexit
        atexit.register(lambda: shutil.rmtree(temp_dir))

        return manager

🔄 Provider Dependencies

Multi-Level Dependencies

class ApplicationModule(Module):
    @provider
    def database_connection(self) -> IDatabaseConnection:
        """Level 1: Basic connection"""
        return PostgresConnection("postgresql://...")

    @provider
    def user_repository(self, db: IDatabaseConnection) -> IUserRepository:
        """Level 2: Depends on connection"""
        return SqlUserRepository(db)

    @provider
    def order_repository(self, db: IDatabaseConnection) -> IOrderRepository:
        """Level 2: Depends on connection"""
        return SqlOrderRepository(db)

    @provider
    def user_service(self, user_repo: IUserRepository, email_svc: IEmailService) -> IUserService:
        """Level 3: Depends on repository and email"""
        return UserService(user_repo, email_svc)

    @provider
    def order_service(self, order_repo: IOrderRepository, payment_svc: IPaymentService) -> IOrderService:
        """Level 3: Depends on repository and payment"""
        return OrderService(order_repo, payment_svc)

Circular Dependency Prevention

# ✅ Good: No circular dependencies
class GoodModule(Module):
    @provider
    def service_a(self, repo: IRepository) -> IServiceA:
        return ServiceA(repo)

    @provider
    def service_b(self, service_a: IServiceA) -> IServiceB:
        return ServiceB(service_a)

# ❌ Bad: Circular dependency
class BadModule(Module):
    @provider
    def service_a(self, service_b: IServiceB) -> IServiceA:
        return ServiceA(service_b)  # Depends on B

    @provider
    def service_b(self, service_a: IServiceA) -> IServiceB:
        return ServiceB(service_a)  # Depends on A

Optional Dependencies

class FlexibleModule(Module):
    @provider
    def notification_service(self, email_svc: Optional[IEmailService] = None) -> INotificationService:
        """Create notification service with optional email"""
        if email_svc:
            return EmailNotificationService(email_svc)
        else:
            return ConsoleNotificationService()

    @provider
    def cache_service(self) -> ICache:
        """Create cache service with fallback"""
        try:
            return RedisCache(host="redis-server")
        except ConnectionError:
            return InMemoryCache()

🧪 Testing with Providers

Provider Testing

def test_provider_creation():
    """Test that providers create correct instances"""
    container = InjectQ()
    container.install(ServiceModule())

    # Test provider-created service
    user_service = container.get(IUserService)
    assert isinstance(user_service, UserService)

    # Test dependencies were injected
    assert user_service.user_repository is not None
    assert user_service.email_service is not None

def test_provider_with_mocks():
    """Test provider with mocked dependencies"""
    container = InjectQ()

    # Mock dependencies
    mock_repo = MockUserRepository()
    mock_email = MockEmailService()

    container.bind(IUserRepository, mock_repo)
    container.bind(IEmailService, mock_email)

    # Install module with providers
    container.install(ServiceModule())

    # Get provider-created service
    user_service = container.get(IUserService)

    # Verify mocks were used
    assert user_service.user_repository is mock_repo
    assert user_service.email_service is mock_email

Provider Override

class TestProvidersModule(Module):
    @provider
    def user_service(self) -> IUserService:
        """Override provider for testing"""
        return MockUserService()

def test_with_provider_override():
    """Test with overridden provider"""
    container = InjectQ()

    # Install production module
    container.install(ServiceModule())

    # Override specific provider
    container.install(TestProvidersModule())

    # Get service
    user_service = container.get(IUserService)

    # Should be mock, not real service
    assert isinstance(user_service, MockUserService)

Provider Dependency Testing

def test_provider_dependencies():
    """Test that provider dependencies are correctly resolved"""
    container = InjectQ()
    container.install(ComplexModule())

    # Get service with complex dependency chain
    payment_processor = container.get(IPaymentProcessor)

    # Verify entire dependency chain
    assert payment_processor.message_queue is not None
    assert payment_processor.database is not None

    # Verify MQ has its dependencies
    mq = payment_processor.message_queue
    assert mq.credentials is not None
    assert mq.retry_policy is not None

🚨 Provider Anti-Patterns

1. Complex Logic in Providers

# ❌ Bad: Too much logic in provider
class BadModule(Module):
    @provider
    def complex_service(self) -> IService:
        # Too much setup logic
        config = self._load_config()
        credentials = self._decrypt_credentials(config)
        connection = self._create_connection(credentials)
        pool = self._create_pool(connection)
        service = self._create_service(pool)

        # Business logic mixed in
        if config.environment == "prod":
            service.enable_monitoring()
        else:
            service.disable_monitoring()

        return service

# ✅ Good: Extract logic to separate methods/classes
class GoodModule(Module):
    def __init__(self, config: AppConfig):
        self.config = config

    @provider
    def service(self) -> IService:
        """Simple provider using factory"""
        return ServiceFactory.create(self.config)

class ServiceFactory:
    @staticmethod
    def create(config: AppConfig) -> IService:
        credentials = CredentialLoader.load(config)
        connection = ConnectionFactory.create(credentials)
        pool = PoolFactory.create(connection, config)
        service = ServiceFactory._create_service(pool, config)

        if config.environment == "prod":
            service.enable_monitoring()

        return service

2. Provider Side Effects

# ❌ Bad: Side effects in provider
class BadModule(Module):
    @provider
    def database_service(self) -> IDatabaseService:
        service = DatabaseService()

        # Side effect: modifies global state
        global_config.database_initialized = True

        # Side effect: creates files
        os.makedirs("/tmp/app_data", exist_ok=True)

        return service

# ✅ Good: Pure providers
class GoodModule(Module):
    @provider
    def database_service(self) -> IDatabaseService:
        return DatabaseService()

    def initialize(self):
        """Call this separately for side effects"""
        global_config.database_initialized = True
        os.makedirs("/tmp/app_data", exist_ok=True)

3. Provider Tight Coupling

# ❌ Bad: Tight coupling in provider
class BadModule(Module):
    @provider
    def user_service(self) -> IUserService:
        # Direct instantiation
        repo = SqlUserRepository(PostgresConnection())
        email = SmtpEmailService()
        return UserService(repo, email)

# ✅ Good: Loose coupling through dependencies
class GoodModule(Module):
    @provider
    def user_service(self, user_repo: IUserRepository, email_svc: IEmailService) -> IUserService:
        return UserService(user_repo, email_svc)

    @provider
    def user_repository(self, db: IDatabaseConnection) -> IUserRepository:
        return SqlUserRepository(db)

    @provider
    def email_service(self) -> IEmailService:
        return SmtpEmailService()

4. Provider Overuse

# ❌ Bad: Provider for everything
class OveruseModule(Module):
    @provider
    def simple_string(self) -> str:
        return "hello"

    @provider
    def simple_number(self) -> int:
        return 42

    @provider
    def simple_list(self) -> List[str]:
        return ["a", "b", "c"]

# ✅ Good: Use providers for complex objects only
class GoodModule(Module):
    @provider
    def complex_service(self, repo: IRepository, config: AppConfig) -> IService:
        return ComplexService(repo, config)

    def configure(self, binder):
        # Simple values can use regular bindings
        binder.bind(str, "hello")
        binder.bind(int, 42)
        binder.bind(List[str], ["a", "b", "c"])

🏆 Best Practices

1. Keep Providers Simple

# ✅ Simple provider
class SimpleModule(Module):
    @provider
    def database_pool(self) -> IDatabasePool:
        return DatabasePool(host="localhost", max_conn=20)

# ✅ Extract complex logic
class ComplexModule(Module):
    @provider
    def payment_processor(self) -> IPaymentProcessor:
        return PaymentProcessorFactory.create(self.config)

2. Use Meaningful Names

# ✅ Good naming
class GoodModule(Module):
    @provider
    def user_notification_service(self) -> IUserNotificationService:
        return EmailUserNotificationService()

    @provider
    def admin_notification_service(self) -> IAdminNotificationService:
        return SmsAdminNotificationService()

# ❌ Bad naming
class BadModule(Module):
    @provider
    def service1(self) -> IService1:
        return Service1Impl()

    @provider
    def svc2(self) -> IService2:
        return Service2Impl()

3. Document Providers

class DocumentedModule(Module):
    @provider
    def user_authentication_service(self, user_repo: IUserRepository, jwt_config: JWTConfig) -> IAuthenticationService:
        """
        Create user authentication service.

        This provider creates an authentication service that handles
        user login, logout, and token validation.

        Args:
            user_repo: Repository for user data access
            jwt_config: Configuration for JWT token handling

        Returns:
            Configured authentication service instance

        Dependencies:
            - IUserRepository: For user data access
            - JWTConfig: For token configuration

        Notes:
            - Uses bcrypt for password hashing
            - Tokens expire after 24 hours
            - Supports refresh token rotation
        """
        return JWTAuthenticationService(user_repo, jwt_config)

4. Handle Errors Gracefully

class RobustModule(Module):
    @provider
    def external_api_client(self) -> IExternalAPI:
        """Create external API client with error handling"""
        try:
            return HttpExternalAPI(
                base_url=os.getenv("API_BASE_URL"),
                api_key=os.getenv("API_KEY"),
                timeout=30
            )
        except (ValueError, ConnectionError) as e:
            # Fallback to mock in case of configuration errors
            logger.warning(f"Failed to create external API client: {e}")
            return MockExternalAPI()

    @provider
    def cache_service(self) -> ICache:
        """Create cache service with fallback"""
        cache_configs = [
            lambda: RedisCache(host=os.getenv("REDIS_HOST")),
            lambda: MemcachedCache(host=os.getenv("MEMCACHED_HOST")),
            lambda: InMemoryCache(),  # Always works
        ]

        for config_func in cache_configs:
            try:
                return config_func()
            except Exception as e:
                logger.warning(f"Failed to create cache: {e}")
                continue

        raise RuntimeError("All cache configurations failed")

5. Test Providers Thoroughly

def test_provider_error_handling():
    """Test provider error handling"""
    # Test with missing environment variables
    with patch.dict(os.environ, {}, clear=True):
        container = InjectQ()
        container.install(RobustModule())

        # Should get fallback mock
        api_client = container.get(IExternalAPI)
        assert isinstance(api_client, MockExternalAPI)

def test_provider_fallback_chain():
    """Test provider fallback chain"""
    container = InjectQ()
    container.install(RobustModule())

    # Should try Redis first
    cache = container.get(ICache)
    # Verify it's the expected type based on configuration

⚡ Advanced Provider Features

Async Providers

class AsyncModule(Module):
    @provider
    async def async_database_pool(self) -> IAsyncDatabasePool:
        """Create async database pool"""
        pool = await AsyncDatabasePool.create(
            host="localhost",
            port=5432,
            database="myapp"
        )
        return pool

    @provider
    async def async_user_service(self, pool: IAsyncDatabasePool) -> IAsyncUserService:
        """Create async user service"""
        return AsyncUserService(pool)

Provider Scopes

class ScopedProvidersModule(Module):
    @provider(scope="singleton")
    def application_config(self) -> IAppConfig:
        """Singleton provider"""
        return AppConfig.from_env()

    @provider(scope="scoped")
    def request_context(self) -> IRequestContext:
        """Scoped provider"""
        return RequestContext()

    @provider(scope="transient")
    def validator(self) -> IValidator:
        """Transient provider"""
        return DataValidator()

Provider with Lifecycle

class LifecycleModule(Module):
    @provider
    def managed_service(self) -> IManagedService:
        """Create service with lifecycle management"""
        service = ManagedService()

        # Register lifecycle hooks
        container.on_shutdown(service.cleanup)

        return service

    @provider
    def health_check_service(self) -> IHealthChecker:
        """Create health checker for all providers"""
        return CompositeHealthChecker([
            DatabaseHealthCheck(),
            CacheHealthCheck(),
            ExternalAPIHealthCheck(),
        ])

🎯 Summary

Provider modules provide:

• Factory functions - Create complex service instances
• Dependency injection - Automatic dependency resolution
• Flexibility - Handle complex creation logic
• Testability - Easy to mock and override
• Clean separation - Separate creation from usage

Key principles: - Keep providers simple and focused - Use meaningful names and documentation - Handle errors gracefully with fallbacks - Test thoroughly including error cases - Avoid side effects and tight coupling

Common patterns: - Complex object creation with dependencies - Conditional providers based on environment - Resource management with cleanup - Multi-level dependency chains - Error handling with fallbacks

Ready to explore module composition?