python-dependency-injector/objects/providers.py

278 lines
6.8 KiB
Python
Raw Normal View History

2015-01-04 17:26:33 +03:00
"""
Standard providers.
"""
from collections import Iterable
2015-01-28 01:48:33 +03:00
from .injections import Injection, InitArg, Attribute, Method
2015-01-10 12:24:25 +03:00
2015-01-04 17:26:33 +03:00
class Provider(object):
"""
Base provider class.
"""
2015-01-11 16:03:45 +03:00
__is_objects_provider__ = True
2015-01-11 19:10:11 +03:00
__overridden_by__ = list()
def __init__(self):
"""
Initializer.
"""
self.__overridden_by__ = list()
2015-01-11 16:03:45 +03:00
2015-01-04 17:26:33 +03:00
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
raise NotImplementedError()
2015-01-11 19:10:11 +03:00
def __override__(self, provider):
"""
Overrides provider with another provider.
"""
self.__overridden_by__.append(provider)
2015-01-11 16:03:45 +03:00
def prepare_injections(injections):
"""
Prepares injections list to injection.
"""
return [(injection.name, injection.value) for injection in injections]
def fetch_injections(injections, injection_type):
"""
Fetches injections of injection type from list.
"""
return tuple([injection
for injection in injections
if isinstance(injection, injection_type)])
2015-01-10 12:24:25 +03:00
2015-01-04 17:26:33 +03:00
class NewInstance(Provider):
"""
New instance providers will create and return new instance on every call.
"""
2015-01-10 12:24:25 +03:00
def __init__(self, provides, *injections):
2015-01-04 17:26:33 +03:00
"""
Initializer.
"""
self.provides = provides
2015-01-11 16:03:45 +03:00
self.init_injections = fetch_injections(injections, InitArg)
self.attribute_injections = fetch_injections(injections, Attribute)
self.method_injections = fetch_injections(injections, Method)
2015-01-11 19:10:11 +03:00
super(NewInstance, self).__init__()
2015-01-04 17:26:33 +03:00
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
2015-01-11 19:10:11 +03:00
if self.__overridden_by__:
return self.__overridden_by__[-1].__call__(*args, **kwargs)
2015-01-11 16:03:45 +03:00
init_injections = prepare_injections(self.init_injections)
init_injections = dict(init_injections)
2015-01-10 12:24:25 +03:00
init_injections.update(kwargs)
2015-01-04 17:26:33 +03:00
2015-01-10 12:24:25 +03:00
provided = self.provides(*args, **init_injections)
2015-01-11 16:03:45 +03:00
attribute_injections = prepare_injections(self.attribute_injections)
for name, injectable in attribute_injections:
2015-01-10 12:24:25 +03:00
setattr(provided, name, injectable)
2015-01-11 16:03:45 +03:00
method_injections = prepare_injections(self.method_injections)
for name, injectable in method_injections:
2015-01-10 12:24:25 +03:00
getattr(provided, name)(injectable)
2015-01-04 17:26:33 +03:00
2015-01-10 12:24:25 +03:00
return provided
2015-01-04 17:26:33 +03:00
class Singleton(NewInstance):
"""
Singleton provider will create instance once and return it on every call.
"""
def __init__(self, *args, **kwargs):
"""
Initializer.
"""
self.instance = None
super(Singleton, self).__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
if not self.instance:
self.instance = super(Singleton, self).__call__(*args, **kwargs)
return self.instance
2015-01-28 01:21:31 +03:00
def _reset_instance(self):
"""
Resets instance.
"""
self.instance = None
class Scoped(Singleton):
"""
Scoped provider will create instance once for every scope and return it on every call.
"""
def __init__(self, *args, **kwargs):
"""
Initializer.
"""
self.is_in_scope = None
super(Scoped, self).__init__(*args, **kwargs)
def in_scope(self):
"""
Sets provider in "in scope" state.
"""
self.is_in_scope = True
self._reset_instance()
def out_of_scope(self):
"""
Sets provider in "out of scope" state.
"""
self.is_in_scope = False
self._reset_instance()
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
if not self.is_in_scope:
raise RuntimeError('Trying to provide {} while provider is not in scope'.format(self.provides))
return super(Scoped, self).__call__(*args, **kwargs)
def __enter__(self):
"""
With __enter__() implementation. Makes provider to be in scope.
"""
self.in_scope()
return self
def __exit__(self, *_):
"""
With __exit__() implementation. Makes provider to be out of scope.
"""
self.out_of_scope()
2015-01-04 17:26:33 +03:00
class ExternalDependency(Provider):
"""
External dependency provider.
"""
def __init__(self, instance_of):
"""
Initializer
:param instance_of: type
"""
if not isinstance(instance_of, Iterable):
instance_of = (instance_of,)
self.instance_of = instance_of
self.dependency = None
super(ExternalDependency, self).__init__()
def satisfy(self, provider):
"""
Satisfies an external dependency.
:param provider: Provider
"""
self.dependency = provider
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
if not self.dependency:
raise ValueError('Dependency is not satisfied')
result = self.dependency.__call__(*args, **kwargs)
if not any((isinstance(result, possible_type) for possible_type in self.instance_of)):
raise TypeError('{} is not an instance of {}'.format(result, self.instance_of))
return result
2015-01-04 17:26:33 +03:00
class _StaticProvider(Provider):
"""
Static provider is base implementation that provides exactly the same as
it got on input.
"""
def __init__(self, provides):
"""
Initializer.
"""
self.provides = provides
2015-01-11 19:10:11 +03:00
super(_StaticProvider, self).__init__()
2015-01-04 17:26:33 +03:00
def __call__(self):
"""
Returns provided instance.
"""
2015-01-11 19:10:11 +03:00
if self.__overridden_by__:
return self.__overridden_by__[-1].__call__()
2015-01-04 17:26:33 +03:00
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.
"""
2015-01-28 01:48:33 +03:00
class Callable(Provider):
"""
Callable providers will provides callable calls with some predefined
dependencies injections.
"""
def __init__(self, calls, *injections):
"""
Initializer.
"""
self.calls = calls
self.injections = fetch_injections(injections, Injection)
super(Callable, self).__init__()
def __call__(self, *args, **kwargs):
"""
Returns provided instance.
"""
if self.__overridden_by__:
return self.__overridden_by__[-1].__call__(*args, **kwargs)
injections = prepare_injections(self.injections)
injections = dict(injections)
injections.update(kwargs)
return self.calls(*args, **injections)