"""Providers module.""" import six from .injections import KwArg from .utils import ensure_is_provider from .utils import is_kwarg_injection from .utils import is_attribute_injection from .utils import is_method_injection from .utils import get_injectable_kwargs from .utils import GLOBAL_LOCK from .errors import Error class Provider(object): """Base provider class.""" __IS_PROVIDER__ = True __slots__ = ('overridden_by',) def __init__(self): """Initializer.""" self.overridden_by = None def __call__(self, *args, **kwargs): """Return provided instance.""" if self.overridden_by: return self.last_overriding(*args, **kwargs) return self._provide(*args, **kwargs) def _provide(self, *args, **kwargs): """Providing strategy implementation. Abstract protected method that implements providing strategy of particular provider. Current method is called every time when not overridden provider is called. Need to be overridden in subclasses. """ raise NotImplementedError() def delegate(self): """Return provider's delegate.""" return Delegate(self) def override(self, provider): """Override provider with another provider.""" if not self.is_overridden: self.overridden_by = (ensure_is_provider(provider),) else: self.overridden_by += (ensure_is_provider(provider),) @property def is_overridden(self): """Check if provider is overridden by another provider.""" return bool(self.overridden_by) @property def last_overriding(self): """Return last overriding provider.""" try: return self.overridden_by[-1] except (TypeError, IndexError): raise Error('Provider {0} is not overridden'.format(str(self))) def reset_last_overriding(self): """Reset last overriding provider.""" if not self.is_overridden: raise Error('Provider {0} is not overridden'.format(str(self))) self.overridden_by = self.overridden_by[:-1] def reset_override(self): """Reset all overriding providers.""" self.overridden_by = None class Delegate(Provider): """Provider's delegate.""" __slots__ = ('delegated',) def __init__(self, delegated): """Initializer. :type delegated: Provider """ self.delegated = ensure_is_provider(delegated) super(Delegate, self).__init__() def _provide(self, *args, **kwargs): """Return provided instance.""" return self.delegated class Factory(Provider): """Factory provider. Factory provider creates new instance of specified class on every call. """ __slots__ = ('provides', 'kwargs', 'attributes', 'methods') def __init__(self, provides, *injections, **kwargs): """Initializer.""" if not callable(provides): raise Error('Factory provider expects to get callable, ' + 'got {0} instead'.format(str(provides))) self.provides = provides self.kwargs = tuple(injection for injection in injections if is_kwarg_injection(injection)) if kwargs: self.kwargs += tuple(KwArg(name, value) for name, value in six.iteritems(kwargs)) self.attributes = tuple(injection for injection in injections if is_attribute_injection(injection)) self.methods = tuple(injection for injection in injections if is_method_injection(injection)) super(Factory, self).__init__() def _provide(self, *args, **kwargs): """Return provided instance.""" instance = self.provides(*args, **get_injectable_kwargs(kwargs, self.kwargs)) for attribute in self.attributes: setattr(instance, attribute.name, attribute.value) for method in self.methods: getattr(instance, method.name)(method.value) return instance @property def injections(self): """Return tuple of all injections.""" return self.kwargs + self.attributes + self.methods class Singleton(Provider): """Singleton provider. Singleton provider will create instance once and return it on every call. """ __slots__ = ('instance', 'factory') def __init__(self, provides, *injections, **kwargs): """Initializer.""" self.instance = None self.factory = Factory(provides, *injections, **kwargs) super(Singleton, self).__init__() def _provide(self, *args, **kwargs): """Return provided instance.""" with GLOBAL_LOCK: if not self.instance: self.instance = self.factory(*args, **kwargs) return self.instance def reset(self): """Reset instance.""" self.instance = None class ExternalDependency(Provider): """External dependency provider. Those provider is used when dependency obviously have to be overridden by the client's code, but it's interface is known. """ __slots__ = ('instance_of',) def __init__(self, instance_of): """Initializer.""" if not isinstance(instance_of, six.class_types): raise Error('ExternalDependency provider expects to get class, ' + 'got {0} instead'.format(str(instance_of))) self.instance_of = instance_of super(ExternalDependency, self).__init__() def __call__(self, *args, **kwargs): """Return provided instance.""" if not self.is_overridden: raise Error('Dependency is not defined') instance = self.last_overriding(*args, **kwargs) if not isinstance(instance, self.instance_of): raise Error('{0} is not an '.format(instance) + 'instance of {0}'.format(self.instance_of)) return instance def provided_by(self, provider): """Set external dependency provider.""" return self.override(provider) class StaticProvider(Provider): """Static provider. Static provider is base implementation that provides exactly the same as it got on input. """ __slots__ = ('provides',) def __init__(self, provides): """Initializer.""" self.provides = provides super(StaticProvider, self).__init__() def _provide(self, *args, **kwargs): """Return provided instance.""" return self.provides class Class(StaticProvider): """Class provider provides class.""" class Object(StaticProvider): """Object provider provides object.""" class Function(StaticProvider): """Function provider provides function.""" class Value(StaticProvider): """Value provider provides value.""" class Callable(Provider): """Callable provider. Callable provider provides callable that is called on every provider call with some predefined dependency injections. """ __slots__ = ('callback', 'kwargs') def __init__(self, callback, **kwargs): """Initializer.""" if not callable(callback): raise Error('Callable expected, got {0}'.format(str(callback))) self.callback = callback self.kwargs = tuple(KwArg(name, value) for name, value in six.iteritems(kwargs)) super(Callable, self).__init__() def _provide(self, *args, **kwargs): """Return provided instance.""" return self.callback(*args, **get_injectable_kwargs(kwargs, self.kwargs)) class Config(Provider): """Config provider. Config provider provides dict values. Also config provider creates child config objects for all undefined attribute calls. It makes possible to create deferred config value provider. """ __slots__ = ('value',) def __init__(self, value=None): """Initializer.""" if not value: value = dict() self.value = value super(Config, self).__init__() def __getattr__(self, item): """Return instance of deferred config.""" return ChildConfig(parents=(item,), root_config=self) def _provide(self, paths=None): """Return provided instance.""" value = self.value if paths: for path in paths: try: value = value[path] except KeyError: raise Error('Config key ' '"{0}" is undefined'.format('.'.join(paths))) return value def update_from(self, value): """Update current value from another one.""" self.value.update(value) class ChildConfig(Provider): """Child config provider. Child config provide an value from the root config object according to the current path in the config tree. """ __slots__ = ('parents', 'root_config') def __init__(self, parents, root_config): """Initializer.""" self.parents = parents self.root_config = root_config super(ChildConfig, self).__init__() def __getattr__(self, item): """Return instance of deferred config.""" return ChildConfig(parents=self.parents + (item,), root_config=self.root_config) def _provide(self, *args, **kwargs): """Return provided instance.""" return self.root_config(self.parents)