Module dexa_sdk.agent.config.injection_context

Injection context implementation.

"""Injection context implementation."""

from collections import namedtuple
import copy
import logging
from typing import Mapping

from ..config.injector import Injector
from aries_cloudagent.config.base import BaseInjector, InjectorError
from aries_cloudagent.config.settings import Settings

Scope = namedtuple("Scope", "name injector")
LOGGER = logging.getLogger(__name__)


class InjectionContextError(InjectorError):
    """Base class for issues in the injection context."""


class InjectionContext(BaseInjector):
    """Manager for configuration settings and class providers."""

    ROOT_SCOPE = "application"

    def __init__(
        self, *, settings: Mapping[str, object] = None, enforce_typing: bool = True
    ):
        """Initialize a `ServiceConfig`."""
        self._injector = Injector(settings, enforce_typing=enforce_typing)
        self._scope_name = InjectionContext.ROOT_SCOPE
        self._scopes = []

    @property
    def injector(self) -> Injector:
        """Accessor for scope-specific injector."""
        return self._injector

    @injector.setter
    def injector(self, injector: Injector):
        """Setter for scope-specific injector."""
        self._injector = injector

    @property
    def scope_name(self) -> str:
        """Accessor for the current scope name."""
        return self._scope_name

    @scope_name.setter
    def scope_name(self, scope_name: str):
        """Accessor for the current scope name."""
        self._scope_name = scope_name

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

    @settings.setter
    def settings(self, settings: Settings):
        """Setter for scope-specific settings."""
        self.injector.settings = settings

    def update_settings(self, settings: Mapping[str, object]):
        """Update the scope with additional settings."""
        if settings:
            self.injector.settings = self.injector.settings.extend(settings)

    def start_scope(
        self, scope_name: str, settings: Mapping[str, object] = None
    ) -> "InjectionContext":
        """Begin a new named scope.

        Args:
            scope_name: The unique name for the scope being entered
            settings: An optional mapping of additional settings to apply

        Returns:
            A new injection context representing the scope

        """
        if not scope_name:
            raise InjectionContextError("Scope name must be non-empty")
        if self.scope_name == scope_name:
            raise InjectionContextError(
                "Cannot re-enter scope: {}".format(scope_name))
        for scope in self._scopes:
            if scope.name == scope_name:
                raise InjectionContextError(
                    "Cannot re-enter scope: {}".format(scope_name)
                )
        result = self.copy()
        result._scopes.append(
            Scope(name=self.scope_name, injector=self.injector))
        result.scope_name = scope_name
        if settings:
            result.update_settings(settings)
        return result

    def injector_for_scope(self, scope_name: str) -> Injector:
        """Fetch the injector for a specific scope.

        Args:
            scope_name: The unique scope identifier
        """
        if scope_name == self.scope_name:
            return self.injector
        for scope in self._scopes:
            if scope.name == scope_name:
                return scope.injector
        return None

    async def inject(
        self,
        base_cls: type,
        settings: Mapping[str, object] = None,
        *,
        required: bool = True
    ) -> object:
        """
        Get the provided instance of a given class identifier.

        Args:
            cls: The base class to retrieve an instance of
            settings: An optional mapping providing configuration to the provider

        Returns:
            An instance of the base class, or None

        """
        return await self.injector.inject(base_cls, settings, required=required)

    def copy(self) -> "InjectionContext":
        """Produce a copy of the injector instance."""
        result = copy.copy(self)
        result._injector = self.injector.copy()
        result._scopes = self._scopes.copy()
        return result

Classes

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

Manager for configuration settings and class providers.

Initialize a ServiceConfig.

Expand source code

class InjectionContext(BaseInjector):
    """Manager for configuration settings and class providers."""

    ROOT_SCOPE = "application"

    def __init__(
        self, *, settings: Mapping[str, object] = None, enforce_typing: bool = True
    ):
        """Initialize a `ServiceConfig`."""
        self._injector = Injector(settings, enforce_typing=enforce_typing)
        self._scope_name = InjectionContext.ROOT_SCOPE
        self._scopes = []

    @property
    def injector(self) -> Injector:
        """Accessor for scope-specific injector."""
        return self._injector

    @injector.setter
    def injector(self, injector: Injector):
        """Setter for scope-specific injector."""
        self._injector = injector

    @property
    def scope_name(self) -> str:
        """Accessor for the current scope name."""
        return self._scope_name

    @scope_name.setter
    def scope_name(self, scope_name: str):
        """Accessor for the current scope name."""
        self._scope_name = scope_name

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

    @settings.setter
    def settings(self, settings: Settings):
        """Setter for scope-specific settings."""
        self.injector.settings = settings

    def update_settings(self, settings: Mapping[str, object]):
        """Update the scope with additional settings."""
        if settings:
            self.injector.settings = self.injector.settings.extend(settings)

    def start_scope(
        self, scope_name: str, settings: Mapping[str, object] = None
    ) -> "InjectionContext":
        """Begin a new named scope.

        Args:
            scope_name: The unique name for the scope being entered
            settings: An optional mapping of additional settings to apply

        Returns:
            A new injection context representing the scope

        """
        if not scope_name:
            raise InjectionContextError("Scope name must be non-empty")
        if self.scope_name == scope_name:
            raise InjectionContextError(
                "Cannot re-enter scope: {}".format(scope_name))
        for scope in self._scopes:
            if scope.name == scope_name:
                raise InjectionContextError(
                    "Cannot re-enter scope: {}".format(scope_name)
                )
        result = self.copy()
        result._scopes.append(
            Scope(name=self.scope_name, injector=self.injector))
        result.scope_name = scope_name
        if settings:
            result.update_settings(settings)
        return result

    def injector_for_scope(self, scope_name: str) -> Injector:
        """Fetch the injector for a specific scope.

        Args:
            scope_name: The unique scope identifier
        """
        if scope_name == self.scope_name:
            return self.injector
        for scope in self._scopes:
            if scope.name == scope_name:
                return scope.injector
        return None

    async def inject(
        self,
        base_cls: type,
        settings: Mapping[str, object] = None,
        *,
        required: bool = True
    ) -> object:
        """
        Get the provided instance of a given class identifier.

        Args:
            cls: The base class to retrieve an instance of
            settings: An optional mapping providing configuration to the provider

        Returns:
            An instance of the base class, or None

        """
        return await self.injector.inject(base_cls, settings, required=required)

    def copy(self) -> "InjectionContext":
        """Produce a copy of the injector instance."""
        result = copy.copy(self)
        result._injector = self.injector.copy()
        result._scopes = self._scopes.copy()
        return result

Ancestors

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

Class variables

var ROOT_SCOPE

Instance variables

var injector : Injector

Accessor for scope-specific injector.

Expand source code

@property
def injector(self) -> Injector:
    """Accessor for scope-specific injector."""
    return self._injector

var scope_name : str

Accessor for the current scope name.

Expand source code

@property
def scope_name(self) -> str:
    """Accessor for the current scope name."""
    return self._scope_name

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.injector.settings

Methods

def copy(self) ‑> InjectionContext

Produce a copy of the injector instance.

Expand source code

def copy(self) -> "InjectionContext":
    """Produce a copy of the injector instance."""
    result = copy.copy(self)
    result._injector = self.injector.copy()
    result._scopes = self._scopes.copy()
    return result

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

Get the provided instance of a given class identifier.

Args

cls

The base class to retrieve an instance of

settings

An optional mapping 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
) -> object:
    """
    Get the provided instance of a given class identifier.

    Args:
        cls: The base class to retrieve an instance of
        settings: An optional mapping providing configuration to the provider

    Returns:
        An instance of the base class, or None

    """
    return await self.injector.inject(base_cls, settings, required=required)

def injector_for_scope(self, scope_name: str) ‑> Injector

Fetch the injector for a specific scope.

Args

scope_name

The unique scope identifier

Expand source code

def injector_for_scope(self, scope_name: str) -> Injector:
    """Fetch the injector for a specific scope.

    Args:
        scope_name: The unique scope identifier
    """
    if scope_name == self.scope_name:
        return self.injector
    for scope in self._scopes:
        if scope.name == scope_name:
            return scope.injector
    return None

def start_scope(self, scope_name: str, settings: Mapping[str, object] = None) ‑> InjectionContext

Begin a new named scope.

Args

scope_name

The unique name for the scope being entered

settings

An optional mapping of additional settings to apply

Returns

A new injection context representing the scope

Expand source code

def start_scope(
    self, scope_name: str, settings: Mapping[str, object] = None
) -> "InjectionContext":
    """Begin a new named scope.

    Args:
        scope_name: The unique name for the scope being entered
        settings: An optional mapping of additional settings to apply

    Returns:
        A new injection context representing the scope

    """
    if not scope_name:
        raise InjectionContextError("Scope name must be non-empty")
    if self.scope_name == scope_name:
        raise InjectionContextError(
            "Cannot re-enter scope: {}".format(scope_name))
    for scope in self._scopes:
        if scope.name == scope_name:
            raise InjectionContextError(
                "Cannot re-enter scope: {}".format(scope_name)
            )
    result = self.copy()
    result._scopes.append(
        Scope(name=self.scope_name, injector=self.injector))
    result.scope_name = scope_name
    if settings:
        result.update_settings(settings)
    return result

def update_settings(self, settings: Mapping[str, object])

Update the scope with additional settings.

Expand source code

def update_settings(self, settings: Mapping[str, object]):
    """Update the scope with additional settings."""
    if settings:
        self.injector.settings = self.injector.settings.extend(settings)

class InjectionContextError (*args, error_code: str = None, **kwargs)

Base class for issues in the injection context.

Initialize a BaseError instance.

Expand source code

class InjectionContextError(InjectorError):
    """Base class for issues in the injection context."""

Ancestors

• aries_cloudagent.config.base.InjectorError
• aries_cloudagent.config.base.ConfigError
• aries_cloudagent.core.error.BaseError
• builtins.Exception
• builtins.BaseException

class Scope (name, injector)

Scope(name, injector)

Ancestors

• builtins.tuple

Instance variables

var injector

Alias for field number 1

var name

Alias for field number 0