File size: 4,619 Bytes
d86b25e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
"""
Service Container for CanRun Dependency Injection
Manages all service dependencies to eliminate circular imports.
"""
import logging
from typing import Dict, Any, Optional, Callable
from abc import ABC, abstractmethod
class ServiceContainer:
"""
Dependency injection container for CanRun services.
Manages service instances and their dependencies.
"""
def __init__(self):
"""Initialize the service container."""
self._services: Dict[str, Any] = {}
self._factories: Dict[str, Callable[[], Any]] = {}
self._singletons: Dict[str, Any] = {}
self.logger = logging.getLogger(__name__)
def register_singleton(self, name: str, factory: Callable[[], Any]) -> None:
"""
Register a singleton service factory.
Args:
name: Service name
factory: Factory function to create the service
"""
self._factories[name] = factory
self.logger.debug(f"Registered singleton service: {name}")
def register_instance(self, name: str, instance: Any) -> None:
"""
Register a service instance directly.
Args:
name: Service name
instance: Service instance
"""
self._services[name] = instance
self.logger.debug(f"Registered service instance: {name}")
def get(self, name: str) -> Any:
"""
Get a service instance by name.
Args:
name: Service name
Returns:
Service instance
Raises:
KeyError: If service is not registered
"""
# Check if instance already exists
if name in self._services:
return self._services[name]
# Check if singleton already created
if name in self._singletons:
return self._singletons[name]
# Create singleton from factory
if name in self._factories:
instance = self._factories[name]()
self._singletons[name] = instance
self.logger.debug(f"Created singleton service: {name}")
return instance
raise KeyError(f"Service '{name}' not registered")
def has(self, name: str) -> bool:
"""
Check if a service is registered.
Args:
name: Service name
Returns:
True if service is registered
"""
return (name in self._services or
name in self._singletons or
name in self._factories)
def clear(self) -> None:
"""Clear all services and factories."""
self._services.clear()
self._factories.clear()
self._singletons.clear()
self.logger.debug("Cleared all services from container")
class ServiceProvider(ABC):
"""
Abstract base class for service providers.
Services can inherit from this to access the container.
"""
def __init__(self, container: ServiceContainer):
"""
Initialize service provider.
Args:
container: Service container instance
"""
self.container = container
self.logger = logging.getLogger(self.__class__.__name__)
@abstractmethod
def initialize(self) -> None:
"""Initialize the service. Must be implemented by subclasses."""
pass
# Global service container instance
_container: Optional[ServiceContainer] = None
def get_container() -> ServiceContainer:
"""
Get the global service container instance.
Returns:
Global service container
"""
global _container
if _container is None:
_container = ServiceContainer()
return _container
def reset_container() -> None:
"""Reset the global service container."""
global _container
_container = None
def inject(service_name: str) -> Callable[[Callable], Callable]:
"""
Decorator to inject a service into a function or method.
Args:
service_name: Name of the service to inject
Returns:
Decorator function
"""
def decorator(func: Callable) -> Callable:
def wrapper(*args, **kwargs):
container = get_container()
service = container.get(service_name)
return func(service, *args, **kwargs)
return wrapper
return decorator |