Source code for aries_cloudagent.config.injector

"""Standard Injector implementation."""

from typing import Mapping, Optional, Type

from .base import BaseProvider, BaseInjector, InjectionError, InjectType
from .provider import InstanceProvider, CachedProvider
from .settings import Settings


[docs]class Injector(BaseInjector): """Injector implementation with static and dynamic bindings.""" def __init__( self, settings: Mapping[str, object] = None, *, enforce_typing: bool = True ): """Initialize an `Injector`.""" self.enforce_typing = enforce_typing self._providers = {} self._settings = Settings(settings) @property def settings(self) -> Settings: """Accessor for scope-specific settings.""" return self._settings @settings.setter def settings(self, settings: Settings): """Setter for scope-specific settings.""" self._settings = settings
[docs] def bind_instance(self, base_cls: Type[InjectType], instance: InjectType): """Add a static instance as a class binding.""" self._providers[base_cls] = InstanceProvider(instance)
[docs] def bind_provider( self, base_cls: Type[InjectType], provider: BaseProvider, *, cache: bool = False ): """Add a dynamic instance resolver as a class binding.""" if not provider: raise ValueError("Class provider binding must be non-empty") if cache and not isinstance(provider, CachedProvider): provider = CachedProvider(provider) self._providers[base_cls] = provider
[docs] def clear_binding(self, base_cls: Type[InjectType]): """Remove a previously-added binding.""" if base_cls in self._providers: del self._providers[base_cls]
[docs] def get_provider(self, base_cls: Type[InjectType]): """Find the provider associated with a class binding.""" return self._providers.get(base_cls)
[docs] def inject_or( self, base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: Optional[InjectType] = None, ) -> Optional[InjectType]: """Get the provided instance of a given class identifier or default if not found. Args: base_cls: The base class to retrieve an instance of settings: An optional dict providing configuration to the provider default: default return value if no instance is found Returns: An instance of the base class, or None """ if not base_cls: raise InjectionError("No base class provided for lookup") provider = self._providers.get(base_cls) if settings: ext_settings = self.settings.extend(settings) else: ext_settings = self.settings if provider: result = provider.provide(ext_settings, self) else: result = None if result and not isinstance(result, base_cls) and self.enforce_typing: raise InjectionError( "Provided instance does not implement the base class: {}".format( base_cls.__name__ ) ) return result if result is not None else default
[docs] def inject( self, base_cls: Type[InjectType], settings: Mapping[str, object] = None, ) -> InjectType: """Get the provided instance of a given class identifier. Args: cls: The base class to retrieve an instance of params: An optional dict providing configuration to the provider Returns: An instance of the base class, or None """ result = self.inject_or(base_cls, settings) if result is None: raise InjectionError( "No instance provided for class: {}".format(base_cls.__name__) ) return result
[docs] def copy(self) -> BaseInjector: """Produce a copy of the injector instance.""" result = Injector(self.settings) result.enforce_typing = self.enforce_typing result._providers = self._providers.copy() return result
def __repr__(self) -> str: """Provide a human readable representation of this object.""" return f"<{self.__class__.__name__}>"