dependency-injection

name	dependency-injection
description	Dependency injection patterns and implementation
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"patterns"}

What I do

• Implement dependency injection
• Design loosely coupled systems
• Create inversion of control containers
• Manage service lifetimes
• Handle circular dependencies
• Configure dependency scopes
• Implement constructor injection
• Use DI for testing

When to use me

When implementing dependency injection or improving testability.

Dependency Injection Basics

from abc import ABC, abstractmethod
from typing import Type, TypeVar, Generic, Dict, Any


T = TypeVar('T')


class DependencyContainer:
    """Simple dependency injection container."""
    
    def __init__(self):
        self._services: Dict[Type, Any] = {}
        self._singletons: Dict[Type, Any] = {}
    
    def register(
        self,
        interface: Type[T],
        implementation: Type[T],
        singleton: bool = False
    ) -> None:
        """Register a dependency."""
        self._services[interface] = (implementation, singleton)
    
    def resolve(self, interface: Type[T]) -> T:
        """Resolve a dependency."""
        if interface not in self._services:
            raise DependencyNotFoundError(interface)
        
        implementation, is_singleton = self._services[interface]
        
        if is_singleton:
            if interface not in self._singletons:
                self._singletons[interface] = implementation()
            return self._singletons[interface]
        
        return implementation()


# Example: Before (tight coupling)
class UserService:
    def __init__(self):
        self.repository = UserRepository()
        self.logger = Logger()
        self.cache = RedisCache()


# Example: After (dependency injection)
class UserService:
    def __init__(
        self,
        repository: UserRepositoryInterface,
        logger: LoggerInterface,
        cache: CacheInterface
    ):
        self.repository = repository
        self.logger = logger
        self.cache = cache

Interface-Based Design

from abc import ABC, abstractmethod
from typing import Protocol, runtimeruntime_checkable
_checkable


@class UserRepositoryProtocol(Protocol):
    """Protocol for user repository."""
    
    @abstractmethod
    def get_by_id(self, user_id: str) -> 'User':
        pass
    
    @abstractmethod
    def get_by_email(self, email: str) -> 'User':
        pass
    
    @abstractmethod
    def save(self, user: 'User') -> None:
        pass


class UserRepository(UserRepositoryProtocol):
    """Concrete implementation of user repository."""
    
    def __init__(self, db_session):
        self.session = db_session
    
    def get_by_id(self, user_id: str) -> 'User':
        return self.session.query(User).get(user_id)
    
    def get_by_email(self, email: str) -> 'User':
        return .session.query(User).filter_by(email=email).first()
    
     () -> :
        .session.add(user)
        .session.commit()


 ():
    
    

     () -> :
        
    

     () -> :
        


 :
    
    
     () -> :
        ()
    
     () -> :
        ()

Constructor Injection

from dataclasses import dataclass


@dataclass
class CreateUserInput:
    """Input for creating a user."""
    email: str
    name: str
    password: str


class UserApplicationService:
    """Application service with dependencies."""
    
    def __init__(
        self,
        user_repository: UserRepositoryProtocol,
        email_service: 'EmailServiceProtocol',
        password_hasher: 'PasswordHasherProtocol',
        logger: LoggerProtocol,
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.password_hasher = password_hasher
        self.logger = logger
    
    def create_user(self, input: CreateUserInput) -> 'User':
        """Create a new user."""
        # Validate
        if self.user_repository.get_by_email(input.email):
            raise UserAlreadyExistsError(input.email)
        
        # Hash password
        hashed_password = self.password_hasher.hash(input.password)
        
        # Create user
        user = User(
            email=input.email,
            name=input.name,
            password_hash=hashed_password,
        )
        
        # Persist
        .user_repository.save(user)
        
        
        .email_service.send_welcome(.email, .name)
        
        
        .logger.info()
        
         user
    
     () -> :
        
        user = .user_repository.get_by_id(user_id)
        
          user:
             UserNotFoundError(user_id)
        
         user

Method and Property Injection

class OrderProcessor:
    """Order processor with method injection."""
    
    def __init__(self):
        self._logger = None
    
    def inject_logger(self, logger: LoggerProtocol) -> None:
        """Method injection for logger."""
        self._logger = logger
    
    @property
    def logger(self) -> LoggerProtocol:
        """Property with lazy initialization."""
        if self._logger is None:
            self._logger = DefaultLogger()
        return self._logger
    
    def process_order(self, order: 'Order') -> None:
        """Process order."""
        self.logger.info(f"Processing order: {order.id}")
        # Process logic...


# Context manager injection
class ServiceScope:
    """Context manager for dependency scope."""
    
    def __init__(self, container: DependencyContainer):
        self.container = container
        self._overrides = {}
    
     ():
         
    
     ():
        
    
     () -> :
        
        ._overrides[interface] = instance
         
    
     () -> T:
        
         interface  ._overrides:
             ._overrides[interface]
         .container.resolve(interface)



 ServiceScope(container)  scope:
    mock_logger = MockLogger()
    
    scope.register_override(LoggerProtocol, mock_logger)
    
    processor = scope.resolve(OrderProcessor)

Advanced Patterns

from typing import Callable, TypeVar, Generic


T = TypeVar('T')


class Factory:
    """Factory pattern for creating instances."""
    
    def __init__(self, container: DependencyContainer):
        self.container = container
    
    def create_user_repository(self) -> UserRepository:
        """Factory method."""
        return UserRepository(
            db_session=self.container.resolve(DatabaseSession)
        )


class LazyProxy(Generic[T]):
    """Lazy proxy for deferred dependency resolution."""
    
    def __init__(self, interface: Type[T]):
        self._interface = interface
        _resolved: T = None
        _instance: T = None
    
    def _get_instance(self) -> T:
        if self._instance is None:
            self._instance = container.resolve(self._interface)
        return self._instance
    
    def __getattr__(self, name: str) -> Any:
        return (._get_instance(), name)
    
     () -> :
         ._get_instance()(*args, **kwargs)



 () -> :
    
     () -> :
         ():
            
            container = get_container()
            
            
            instance = container.resolve(interface)
            
            
             func(instance, *args, **kwargs)
        
         wrapper
     decorator




 ():
    
    logger.info()

Lifecycle Management

from enum import Enum
from dataclasses import dataclass
from typing import Any, Callable


class ServiceLifetime(Enum):
    """Service lifetime options."""
    TRANSIENT = "transient"      # New instance each time
    SCOPED = "scoped"             # One instance per scope
    SINGLETON = "singleton"       # One instance globally


@dataclass
class ServiceDescriptor:
    """Service registration descriptor."""
    interface: Type
    implementation: Type
    lifetime: ServiceLifetime


class ServiceCollection:
    """Service registration collection."""
    
    def __init__(self):
        self._descriptors: list[ServiceDescriptor] = []
    
    def add(
        self,
        interface: Type[T],
        implementation: Type[T],
        lifetime: ServiceLifetime = ServiceLifetime.TRANSIENT
    ) -> 'ServiceCollection':
        """Add service registration."""
        self._descriptors.append(
            ServiceDescriptor(interface, implementation, lifetime)
        )
        return self
    
    def add_transient() -> :
        
         .add(interface, implementation, ServiceLifetime.TRANSIENT)
    
     () -> :
        
         .add(interface, implementation, ServiceLifetime.SCOPED)
    
     () -> :
        
         .add(interface, implementation, ServiceLifetime.SINGLETON)


 :
    
    
     ():
        .collection = collection
        ._singletons: [, ] = {}
        ._scoped: [, [, ]] = {}
        ._current_scope_id = 
    
     () -> :
        
        scope_id = (())
        ._scoped[scope_id] = {}
        
         ServiceScope(, scope_id)
    
     () -> T:
        
        descriptor = ._find_descriptor(interface)
        
          descriptor:
             DependencyNotFoundError(interface)
        
         ._create_instance(descriptor)
    
     () -> ServiceDescriptor:
        
         desc  ._descriptors:
             desc.interface == interface:
                 desc
         
    
     () -> :
        
         descriptor.lifetime == ServiceLifetime.SINGLETON:
             ._get_singleton(descriptor)
         descriptor.lifetime == ServiceLifetime.SCOPED:
             ._get_scoped(descriptor)
        :  
             ._create_impl(descriptor.implementation)
    
     () -> :
        
         descriptor.interface   ._singletons:
            ._singletons[descriptor.interface] = (
                ._create_impl(descriptor.implementation)
            )
         ._singletons[descriptor.interface]
    
     () -> :
        
         ._current_scope_id  :
             NoActiveScopeError()
        
        scope = ._scoped[._current_scope_id]
        
         descriptor.interface   scope:
            scope[descriptor.interface] = (
                ._create_impl(descriptor.implementation)
            )
         scope[descriptor.interface]
    
     () -> :
        
        
        constructor = implementation.__init__
        
        
         inspect
        sig = inspect.signature(constructor)
        
        
        args = []
         param_name, param  sig.parameters.items():
             param_name == :
                
             param.annotation != param.empty:
                instance = .resolve(param.annotation)
                args.append(instance)
             param.default == param.empty:
                 MissingDependencyError(param_name)
        
         constructor(*args)

Best Practices

Dependency Injection Best Practices:

1. Use constructor injection
   Most common pattern
   Makes dependencies explicit

2. Program to interfaces
   Define contracts
   Don't depend on implementations

3. Use appropriate lifetime
   Singleton for stateless services
   Scoped for user-specific
   Transient for stateless operations

4. Avoid service locator
   Don't use container as service locator
   Makes dependencies implicit

5. Handle disposal
   Implement IDisposable when needed
   Clean up resources

6. Avoid circular dependencies
   Refactor to break cycles
   Use property injection

7. Use DI container
   Don't roll your own in production
   Use established frameworks

8. Test with mocks
   Easy to substitute implementations
   Improves testability

9. Configure at composition root
   Configure once at startup
   Keep composition simple

10. Consider lazy initialization
     Defer expensive creation
     Use lazy<T> when appropriate