Module dexa_sdk.agent.config.injector

Standard Injector implementation.

"""Standard Injector implementation."""

from typing import Mapping

from aries_cloudagent.config.base import BaseProvider, BaseInjector, InjectorError
from aries_cloudagent.config.provider import InstanceProvider, CachedProvider
from aries_cloudagent.config.settings import Settings


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

    def bind_instance(self, base_cls: type, instance: object):
        """Add a static instance as a class binding."""
        self._providers[base_cls] = InstanceProvider(instance)

    def bind_provider(
        self, base_cls: type, 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

    def clear_binding(self, base_cls: type):
        """Remove a previously-added binding."""
        if base_cls in self._providers:
            del self._providers[base_cls]

    def get_provider(self, base_cls: type):
        """Find the provider associated with a class binding."""
        return self._providers.get(base_cls)

    async def inject(
        self,
        base_cls: type,
        settings: Mapping[str, object] = None,
        *,
        required: bool = True,
    ):
        """
        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

        """
        if not base_cls:
            raise InjectorError("No base class provided for lookup")
        provider = self._providers.get(base_cls)
        ext_settings = self.settings.extend(
            settings) if settings else self.settings
        if provider:
            result = await provider.provide(ext_settings, self)
        else:
            result = None
        if result is None:
            if required:
                raise InjectorError(
                    "No instance provided for class: {}".format(
                        base_cls.__name__)
                )
        elif not isinstance(result, base_cls) and self.enforce_typing:
            raise InjectorError(
                "Provided instance does not implement the base class: {}".format(
                    base_cls.__name__
                )
            )
        return result

    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__}>"

Classes

class Injector (settings: Mapping[str, object] = None, enforce_typing: bool = True)

Injector implementation with static and dynamic bindings.

Initialize an Injector.

Expand source code

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

    def bind_instance(self, base_cls: type, instance: object):
        """Add a static instance as a class binding."""
        self._providers[base_cls] = InstanceProvider(instance)

    def bind_provider(
        self, base_cls: type, 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

    def clear_binding(self, base_cls: type):
        """Remove a previously-added binding."""
        if base_cls in self._providers:
            del self._providers[base_cls]

    def get_provider(self, base_cls: type):
        """Find the provider associated with a class binding."""
        return self._providers.get(base_cls)

    async def inject(
        self,
        base_cls: type,
        settings: Mapping[str, object] = None,
        *,
        required: bool = True,
    ):
        """
        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

        """
        if not base_cls:
            raise InjectorError("No base class provided for lookup")
        provider = self._providers.get(base_cls)
        ext_settings = self.settings.extend(
            settings) if settings else self.settings
        if provider:
            result = await provider.provide(ext_settings, self)
        else:
            result = None
        if result is None:
            if required:
                raise InjectorError(
                    "No instance provided for class: {}".format(
                        base_cls.__name__)
                )
        elif not isinstance(result, base_cls) and self.enforce_typing:
            raise InjectorError(
                "Provided instance does not implement the base class: {}".format(
                    base_cls.__name__
                )
            )
        return result

    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__}>"

Ancestors

• aries_cloudagent.config.base.BaseInjector
• abc.ABC

Instance variables

var settings : aries_cloudagent.config.settings.Settings

Accessor for scope-specific settings.

Expand source code

@property
def settings(self) -> Settings:
    """Accessor for scope-specific settings."""
    return self._settings

Methods

def bind_instance(self, base_cls: type, instance: object)

Add a static instance as a class binding.

Expand source code

def bind_instance(self, base_cls: type, instance: object):
    """Add a static instance as a class binding."""
    self._providers[base_cls] = InstanceProvider(instance)

def bind_provider(self, base_cls: type, provider: aries_cloudagent.config.base.BaseProvider, *, cache: bool = False)

Add a dynamic instance resolver as a class binding.

Expand source code

def bind_provider(
    self, base_cls: type, 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

def clear_binding(self, base_cls: type)

Remove a previously-added binding.

Expand source code

def clear_binding(self, base_cls: type):
    """Remove a previously-added binding."""
    if base_cls in self._providers:
        del self._providers[base_cls]

def copy(self) ‑> aries_cloudagent.config.base.BaseInjector

Produce a copy of the injector instance.

Expand source code

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 get_provider(self, base_cls: type)

Find the provider associated with a class binding.

Expand source code

def get_provider(self, base_cls: type):
    """Find the provider associated with a class binding."""
    return self._providers.get(base_cls)

async def inject(self, base_cls: type, settings: Mapping[str, object] = None, *, required: bool = True)

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

Expand source code

async def inject(
    self,
    base_cls: type,
    settings: Mapping[str, object] = None,
    *,
    required: bool = True,
):
    """
    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

    """
    if not base_cls:
        raise InjectorError("No base class provided for lookup")
    provider = self._providers.get(base_cls)
    ext_settings = self.settings.extend(
        settings) if settings else self.settings
    if provider:
        result = await provider.provide(ext_settings, self)
    else:
        result = None
    if result is None:
        if required:
            raise InjectorError(
                "No instance provided for class: {}".format(
                    base_cls.__name__)
            )
    elif not isinstance(result, base_cls) and self.enforce_typing:
        raise InjectorError(
            "Provided instance does not implement the base class: {}".format(
                base_cls.__name__
            )
        )
    return result