ets-labs/python-dependency-injector
1
1
Fork 0
mirror of https://github.com/ets-labs/python-dependency-injector.git synced 2025-11-04 09:57:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update providers package structure

Browse Source
This commit is contained in:
Roman Mogilatov 2016-11-04 18:35:53 +02:00
parent e68df7167f
commit 38bdc4f04f
20 changed files with 667 additions and 972 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
setup.py
View File

@ -47,6 +47,26 @@ setup(name='dependency-injector',
'': 'src',
},
ext_modules=[
Extension('dependency_injector.providers.base',
['src/dependency_injector/providers/base.c'],
define_macros=defined_macros,
extra_compile_args=['-O2']),
Extension('dependency_injector.providers.callables',
['src/dependency_injector/providers/callables.c'],
define_macros=defined_macros,
extra_compile_args=['-O2']),
Extension('dependency_injector.providers.factories',
['src/dependency_injector/providers/factories.c'],
define_macros=defined_macros,
extra_compile_args=['-O2']),
Extension('dependency_injector.providers.singletons',
['src/dependency_injector/providers/singletons.c'],
define_macros=defined_macros,
extra_compile_args=['-O2']),
Extension('dependency_injector.providers.static',
['src/dependency_injector/providers/static.c'],
define_macros=defined_macros,
extra_compile_args=['-O2']),
Extension('dependency_injector.providers.injections',
['src/dependency_injector/providers/injections.c'],
define_macros=defined_macros,

61
src/dependency_injector/providers/__init__.py
View File

@ -2,46 +2,49 @@
from .base import (
Provider,
Delegate,
Object,
ExternalDependency,
OverridingContext,
override,
)
from .callable import (
from .callables import (
Callable,
DelegatedCallable,
)
from .creational import (
from .factories import (
Factory,
DelegatedFactory,
)
from .singletons import (
BaseSingleton,
Singleton,
DelegatedSingleton,
ThreadSafeSingleton,
DelegatedThreadSafeSingleton,
ThreadLocalSingleton,
DelegatedThreadLocalSingleton,
)
from .utils import (
GLOBAL_LOCK,
is_provider,
ensure_is_provider,
is_delegated,
represent_provider,
from .static import (
Object,
Delegate,
ExternalDependency,
)
from .injections import (
Injection,
PositionalInjection,
NamedInjection,
)
from .utils import (
GLOBAL_LOCK,
OverridingContext,
is_provider,
ensure_is_provider,
is_delegated,
represent_provider,
)
__all__ = (
'Provider',
'Delegate',
'Object',
'ExternalDependency',
'OverridingContext',
'override',
'Callable',
'DelegatedCallable',
@ -49,19 +52,29 @@ __all__ = (
'Factory',
'DelegatedFactory',
'BaseSingleton',
'Singleton',
'DelegatedSingleton',
'ThreadSafeSingleton',
'DelegatedThreadSafeSingleton',
'ThreadLocalSingleton',
'DelegatedThreadLocalSingleton',
'GLOBAL_LOCK',
'is_provider',
'ensure_is_provider',
'is_delegated',
'represent_provider',
'Object',
'Delegate',
'ExternalDependency',
'Injection',
'PositionalInjection',
'NamedInjection',
'GLOBAL_LOCK',
'OverridingContext',
'is_provider',
'ensure_is_provider',
'is_delegated',
'represent_provider',
)

12
src/dependency_injector/providers/base.pxd Normal file
View File

@ -0,0 +1,12 @@
"""Dependency injector base providers.
Powered by Cython.
"""
cdef class Provider(object):
cdef tuple __overridden
cdef int __overridden_len
cpdef object _provide(self, tuple args, dict kwargs)
cpdef object _call_last_overriding(self, tuple args, dict kwargs)

416
src/dependency_injector/providers/base.py
View File

@ -1,416 +0,0 @@
"""Dependency injector base providers."""
import six
from dependency_injector.errors import Error
from .utils import (
is_provider,
ensure_is_provider,
represent_provider,
)
@six.python_2_unicode_compatible
class Provider(object):
"""Base provider class.
:py:class:`Provider` is callable (implements ``__call__`` method). Every
call to provider object returns provided result, according to the providing
strategy of particular provider. This ``callable`` functionality is a
regular part of providers API and it should be the same for all provider's
subclasses.
Implementation of particular providing strategy should be done in
:py:meth:`Provider._provide` of :py:class:`Provider` subclass. Current
method is called every time when not overridden provider is called.
:py:class:`Provider` implements provider overriding logic that should be
also common for all providers:
.. code-block:: python
provider1 = Factory(SomeClass)
provider2 = Factory(ChildSomeClass)
provider1.override(provider2)
some_instance = provider1()
assert isinstance(some_instance, ChildSomeClass)
Also :py:class:`Provider` implements helper function for creating its
delegates:
.. code-block:: python
provider = Factory(object)
delegate = provider.delegate()
delegated = delegate()
assert provider is delegated
All providers should extend this class.
.. py:attribute:: overridden
Tuple of overriding providers, if any.
:type: tuple[:py:class:`Provider`] | None
"""
__IS_PROVIDER__ = True
__OPTIMIZED_CALLS__ = True
__slots__ = ('overridden', 'provide', '__call__')
def __init__(self):
"""Initializer."""
self.overridden = tuple()
super(Provider, self).__init__()
# Enable __call__() / _provide() optimization
if self.__class__.__OPTIMIZED_CALLS__:
self.__call__ = self.provide = self._provide
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 _call_last_overriding(self, *args, **kwargs):
"""Call last overriding provider and return result."""
return (self.overridden[-1](*args, **kwargs)
if self.overridden
else None)
def provide_injection(self):
"""Injection strategy implementation.
:rtype: object
"""
return self.provide()
def override(self, provider):
"""Override provider with another provider.
:param provider: Overriding provider.
:type provider: :py:class:`Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:return: Overriding provider.
:rtype: :py:class:`Provider`
"""
if provider is self:
raise Error('Provider {0} could not be overridden '
'with itself'.format(self))
if not is_provider(provider):
provider = Object(provider)
self.overridden += (ensure_is_provider(provider),)
# Disable __call__() / _provide() optimization
if self.__class__.__OPTIMIZED_CALLS__:
self.__call__ = self.provide = self._call_last_overriding
return OverridingContext(self, provider)
def reset_last_overriding(self):
"""Reset last overriding provider.
:raise: :py:exc:`dependency_injector.errors.Error` if provider is not
overridden.
:rtype: None
"""
if not self.overridden:
raise Error('Provider {0} is not overridden'.format(str(self)))
self.overridden = self.overridden[:-1]
if not self.overridden:
# Enable __call__() / _provide() optimization
if self.__class__.__OPTIMIZED_CALLS__:
self.__call__ = self.provide = self._provide
def reset_override(self):
"""Reset all overriding providers.
:rtype: None
"""
self.overridden = tuple()
# Enable __call__() / _provide() optimization
if self.__class__.__OPTIMIZED_CALLS__:
self.__call__ = self.provide = self._provide
def delegate(self):
"""Return provider's delegate.
:rtype: :py:class:`Delegate`
"""
return Delegate(self)
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=None)
__repr__ = __str__
@six.python_2_unicode_compatible
class Delegate(Provider):
""":py:class:`Delegate` provider delegates another provider.
.. code-block:: python
provider = Factory(object)
delegate = Delegate(provider)
delegated = delegate()
assert provider is delegated
.. py:attribute:: delegated
Delegated provider.
:type: :py:class:`Provider`
"""
__slots__ = ('delegated',)
def __init__(self, delegated):
"""Initializer.
:provider delegated: Delegated provider.
:type delegated: :py:class:`Provider`
"""
self.delegated = ensure_is_provider(delegated)
super(Delegate, self).__init__()
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
return self.delegated
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.delegated)
__repr__ = __str__
@six.python_2_unicode_compatible
class Object(Provider):
""":py:class:`Object` provider returns provided instance "as is".
.. py:attribute:: provides
Value that have to be provided.
:type: object
"""
__slots__ = ('provides',)
def __init__(self, provides):
"""Initializer.
:param provides: Value that have to be provided.
:type provides: object
"""
self.provides = provides
super(Object, self).__init__()
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
return self.provides
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.provides)
__repr__ = __str__
@six.python_2_unicode_compatible
class ExternalDependency(Provider):
""":py:class:`ExternalDependency` provider describes dependency interface.
This provider is used for description of dependency interface. That might
be useful when dependency could be provided in the client's code only,
but it's interface is known. Such situations could happen when required
dependency has non-determenistic list of dependencies itself.
.. code-block:: python
database_provider = ExternalDependency(sqlite3.dbapi2.Connection)
database_provider.override(Factory(sqlite3.connect, ':memory:'))
database = database_provider()
.. py:attribute:: instance_of
Class of required dependency.
:type: type
"""
__OPTIMIZED_CALLS__ = False
__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
self.provide = self.__call__
super(ExternalDependency, self).__init__()
def __call__(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: object
"""
if not self.overridden:
raise Error('Dependency is not defined')
instance = self._call_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.
:param provider: Provider that provides required dependency.
:type provider: :py:class:`Provider`
:rtype: None
"""
return self.override(provider)
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.instance_of)
__repr__ = __str__
class OverridingContext(object):
"""Provider overriding context.
:py:class:`OverridingContext` is used by :py:meth:`Provider.override` for
implemeting ``with`` contexts. When :py:class:`OverridingContext` is
closed, overriding that was created in this context is dropped also.
.. code-block:: python
with provider.override(another_provider):
assert provider.overridden
assert not provider.overridden
"""
def __init__(self, overridden, overriding):
"""Initializer.
:param overridden: Overridden provider.
:type overridden: :py:class:`Provider`
:param overriding: Overriding provider.
:type overriding: :py:class:`Provider`
"""
self.overridden = overridden
self.overriding = overriding
def __enter__(self):
"""Do nothing."""
return self.overriding
def __exit__(self, *_):
"""Exit overriding context."""
self.overridden.reset_last_overriding()
def override(overridden):
"""Decorator for overriding providers.
This decorator overrides ``overridden`` provider by decorated one.
.. code-block:: python
@Factory
class SomeClass(object):
pass
@override(SomeClass)
@Factory
class ExtendedSomeClass(SomeClass.cls):
pass
:param overridden: Provider that should be overridden.
:type overridden: :py:class:`Provider`
:return: Overriding provider.
:rtype: :py:class:`Provider`
"""
def decorator(overriding):
overridden.override(overriding)
return overriding
return decorator
def _parse_positional_injections(args):
return tuple(arg if is_provider(arg) else Object(arg)
for arg in args)
def _parse_keyword_injections(kwargs):
return dict((name, arg if is_provider(arg) else Object(arg))
for name, arg in six.iteritems(kwargs))

169
src/dependency_injector/providers/base.pyx Normal file
View File

@ -0,0 +1,169 @@
"""Dependency injector base providers.
Powered by Cython.
"""
cimport cython
from dependency_injector.errors import Error
from .static cimport (
Object,
Delegate,
)
from .utils cimport (
is_provider,
ensure_is_provider,
represent_provider,
OverridingContext,
)
cdef class Provider(object):
"""Base provider class.
:py:class:`Provider` is callable (implements ``__call__`` method). Every
call to provider object returns provided result, according to the providing
strategy of particular provider. This ``callable`` functionality is a
regular part of providers API and it should be the same for all provider's
subclasses.
Implementation of particular providing strategy should be done in
:py:meth:`Provider._provide` of :py:class:`Provider` subclass. Current
method is called every time when not overridden provider is called.
:py:class:`Provider` implements provider overriding logic that should be
also common for all providers:
.. code-block:: python
provider1 = Factory(SomeClass)
provider2 = Factory(ChildSomeClass)
provider1.override(provider2)
some_instance = provider1()
assert isinstance(some_instance, ChildSomeClass)
Also :py:class:`Provider` implements helper function for creating its
delegates:
.. code-block:: python
provider = Factory(object)
delegate = provider.delegate()
delegated = delegate()
assert provider is delegated
All providers should extend this class.
.. py:attribute:: overridden
Tuple of overriding providers, if any.
:type: tuple[:py:class:`Provider`] | None
"""
__IS_PROVIDER__ = True
def __init__(self):
"""Initializer."""
self.__overridden = tuple()
self.__overridden_len = 0
def __call__(self, *args, **kwargs):
"""Return provided object.
Callable interface implementation.
"""
if self.__overridden_len != 0:
return self._call_last_overriding(args, kwargs)
return self._provide(args, kwargs)
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=None)
def __repr__(self):
"""Return string representation of provider.
:rtype: str
"""
return self.__str__()
def override(self, Provider provider):
"""Override provider with another provider.
:param provider: Overriding provider.
:type provider: :py:class:`Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:return: Overriding context.
:rtype: :py:class:`OverridingContext`
"""
if provider is self:
raise Error('Provider {0} could not be overridden '
'with itself'.format(self))
if not is_provider(provider):
provider = Object(provider)
self.__overridden += tuple(ensure_is_provider(provider),)
self.__overridden_len += 1
return OverridingContext(self, provider)
@cython.boundscheck(False)
@cython.wraparound(False)
def reset_last_overriding(self):
"""Reset last overriding provider.
:raise: :py:exc:`dependency_injector.errors.Error` if provider is not
overridden.
:rtype: None
"""
if self.__overridden_len == 0:
raise Error('Provider {0} is not overridden'.format(str(self)))
self.__overridden = self.overridden[:self.__overridden_len - 1]
self.__overridden_len -= 1
def reset_override(self):
"""Reset all overriding providers.
:rtype: None
"""
self.__overridden = tuple()
self.__overridden_len += 0
def delegate(self):
"""Return provider's delegate.
:rtype: :py:class:`Delegate`
"""
return Delegate(self)
cpdef object _provide(self, tuple args, dict 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()
@cython.boundscheck(False)
@cython.wraparound(False)
cpdef object _call_last_overriding(self, tuple args, dict kwargs):
"""Call last overriding provider and return result."""
if self.__overridden_len == 0:
return None
return <object>self.__overridden[self.__overridden_len - 1](*args,
**kwargs)

127
src/dependency_injector/providers/callable.py
View File

@ -1,127 +0,0 @@
"""Dependency injector callable providers."""
import six
from dependency_injector.providers.base import (
Provider,
_parse_positional_injections,
_parse_keyword_injections,
)
from .utils import represent_provider
from dependency_injector.errors import Error
@six.python_2_unicode_compatible
class Callable(Provider):
r""":py:class:`Callable` provider calls wrapped callable on every call.
:py:class:`Callable` supports positional and keyword argument injections:
.. code-block:: python
some_function = Callable(some_function,
'positional_arg1', 'positional_arg2',
keyword_argument1=3, keyword_argument=4)
# or
some_function = Callable(some_function) \
.add_args('positional_arg1', 'positional_arg2') \
.add_kwargs(keyword_argument1=3, keyword_argument=4)
# or
some_function = Callable(some_function)
some_function.add_args('positional_arg1', 'positional_arg2')
some_function.add_kwargs(keyword_argument1=3, keyword_argument=4)
"""
__slots__ = ('provides', 'args', 'kwargs')
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Wrapped callable.
:type provides: callable
"""
if not callable(provides):
raise Error('Provider {0} expected to get callable, '
'got {0}'.format('.'.join((self.__class__.__module__,
self.__class__.__name__)),
provides))
self.provides = provides
self.args = tuple()
self.kwargs = dict()
self.add_args(*args)
self.add_kwargs(**kwargs)
super(Callable, self).__init__()
def add_args(self, *args):
"""Add postional argument injections.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.args += _parse_positional_injections(args)
return self
def add_kwargs(self, **kwargs):
"""Add keyword argument injections.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.kwargs.update(_parse_keyword_injections(kwargs))
return self
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
if self.args:
args = tuple(arg.provide_injection() for arg in self.args) + args
for name, arg in six.iteritems(self.kwargs):
if name not in kwargs:
kwargs[name] = arg.provide_injection()
return self.provides(*args, **kwargs)
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.provides)
__repr__ = __str__
class DelegatedCallable(Callable):
""":py:class:`DelegatedCallable` is a delegated :py:class:`Callable`.
:py:class:`DelegatedCallable` is a :py:class:`Callable`, that is injected
"as is".
"""
def provide_injection(self):
"""Injection strategy implementation.
:rtype: object
"""
return self

14
src/dependency_injector/providers/callables.pxd Normal file
View File

@ -0,0 +1,14 @@
"""Dependency injector callable providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class Callable(Provider):
pass
cdef class DelegatedCallable(Callable):
pass

14
src/dependency_injector/providers/callables.pyx Normal file
View File

@ -0,0 +1,14 @@
"""Dependency injector callable providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class Callable(Provider):
pass
cdef class DelegatedCallable(Callable):
__IS_DELEGATED__ = True

342
src/dependency_injector/providers/creational.py
View File

@ -1,342 +0,0 @@
"""Dependency injector creational providers."""
import threading
import six
from dependency_injector.providers.callable import Callable
from dependency_injector.providers.base import _parse_keyword_injections
from .utils import GLOBAL_LOCK
from dependency_injector.errors import Error
class Factory(Callable):
r""":py:class:`Factory` provider creates new instance on every call.
:py:class:`Factory` supports positional & keyword argument injections,
as well as attribute injections.
Positional and keyword argument injections could be defined like this:
.. code-block:: python
factory = Factory(SomeClass,
'positional_arg1', 'positional_arg2',
keyword_argument1=3, keyword_argument=4)
# or
factory = Factory(SomeClass) \
.add_args('positional_arg1', 'positional_arg2') \
.add_kwargs(keyword_argument1=3, keyword_argument=4)
# or
factory = Factory(SomeClass)
factory.add_args('positional_arg1', 'positional_arg2')
factory.add_kwargs(keyword_argument1=3, keyword_argument=4)
Attribute injections are defined by using :py:meth:`Factory.attributes`:
.. code-block:: python
factory = Factory(SomeClass) \
.add_attributes(attribute1=1, attribute2=2)
Retrieving of provided instance can be performed via calling
:py:class:`Factory` object:
.. code-block:: python
factory = Factory(SomeClass)
some_object = factory()
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
provided_type = None
__slots__ = ('cls', 'attributes')
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Class or other callable that provides object
for creation.
:type provides: type | callable
"""
if (self.__class__.provided_type and
not issubclass(provides, self.__class__.provided_type)):
raise Error('{0} can provide only {1} instances'.format(
self.__class__, self.__class__.provided_type))
self.attributes = dict()
super(Factory, self).__init__(provides, *args, **kwargs)
self.cls = self.provides
def add_attributes(self, **kwargs):
"""Add attribute injections.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.attributes.update(_parse_keyword_injections(kwargs))
return self
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
instance = super(Factory, self)._provide(*args, **kwargs)
for name, arg in six.iteritems(self.attributes):
setattr(instance, name, arg.provide_injection())
return instance
class DelegatedFactory(Factory):
""":py:class:`Factory` that is injected "as is".
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
def provide_injection(self):
"""Injection strategy implementation.
:rtype: object
"""
return self
class Singleton(Factory):
""":py:class:`Singleton` provider returns same instance on every call.
:py:class:`Singleton` provider creates instance once and returns it on
every call. :py:class:`Singleton` extends :py:class:`Factory`, so, please
follow :py:class:`Factory` documentation for getting familiar with
injections syntax.
:py:class:`Singleton` is thread-safe and could be used in multithreading
environment without any negative impact.
Retrieving of provided instance can be performed via calling
:py:class:`Singleton` object:
.. code-block:: python
singleton = Singleton(SomeClass)
some_object = singleton()
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
__slots__ = ('instance',)
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Class or other callable that provides object
for creation.
:type provides: type | callable
"""
self.instance = None
super(Singleton, self).__init__(provides, *args, **kwargs)
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
self.instance = None
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
with GLOBAL_LOCK:
if self.instance is None:
self.instance = super(Singleton, self)._provide(*args,
**kwargs)
return self.instance
class DelegatedSingleton(Singleton):
""":py:class:`Singleton` that is injected "as is".
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
def provide_injection(self):
"""Injection strategy implementation.
:rtype: object
"""
return self
class ThreadLocalSingleton(Factory):
""":py:class:`ThreadLocalSingleton` is singleton based on thread locals.
:py:class:`ThreadLocalSingleton` provider creates instance once for each
thread and returns it on every call. :py:class:`ThreadLocalSingleton`
extends :py:class:`Factory`, so, please follow :py:class:`Factory`
documentation for getting familiar with injections syntax.
:py:class:`ThreadLocalSingleton` is thread-safe and could be used in
multithreading environment without any negative impact.
Retrieving of provided instance can be performed via calling
:py:class:`ThreadLocalSingleton` object:
.. code-block:: python
singleton = ThreadLocalSingleton(SomeClass)
some_object = singleton()
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
__slots__ = ('local_storage',)
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Class or other callable that provides object
for creation.
:type provides: type | callable
"""
self.local_storage = threading.local()
super(ThreadLocalSingleton, self).__init__(provides, *args, **kwargs)
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
self.local_storage.instance = None
def _provide(self, *args, **kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
try:
instance = self.local_storage.instance
except AttributeError:
instance = super(ThreadLocalSingleton, self)._provide(*args,
**kwargs)
self.local_storage.instance = instance
finally:
return instance
class DelegatedThreadLocalSingleton(ThreadLocalSingleton):
""":py:class:`ThreadLocalSingleton` that is injected "as is".
.. py:attribute:: provided_type
If provided type is defined, provider checks that providing class is
its subclass.
:type: type | None
.. py:attribute:: cls
Class that provides object.
Alias for :py:attr:`provides`.
:type: type
"""
def provide_injection(self):
"""Injection strategy implementation.
:rtype: object
"""
return self

14
src/dependency_injector/providers/factories.pxd Normal file
View File

@ -0,0 +1,14 @@
"""Dependency injector factory providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class Factory(Provider):
pass
cdef class DelegatedFactory(Factory):
pass

14
src/dependency_injector/providers/factories.pyx Normal file
View File

@ -0,0 +1,14 @@
"""Dependency injector factory providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class Factory(Provider):
pass
cdef class DelegatedFactory(Factory):
__IS_DELEGATED__ = True

34
src/dependency_injector/providers/singletons.pxd Normal file
View File

@ -0,0 +1,34 @@
"""Dependency injector singleton providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class BaseSingleton(Provider):
pass
cdef class Singleton(BaseSingleton):
pass
cdef class DelegatedSingleton(Singleton):
pass
cdef class ThreadSafeSingleton(Singleton):
pass
cdef class DelegatedThreadSafeSingleton(ThreadSafeSingleton):
pass
cdef class ThreadLocalSingleton(BaseSingleton):
pass
cdef class DelegatedThreadLocalSingleton(ThreadLocalSingleton):
pass

34
src/dependency_injector/providers/singletons.pyx Normal file
View File

@ -0,0 +1,34 @@
"""Dependency injector singleton providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class BaseSingleton(Provider):
pass
cdef class Singleton(BaseSingleton):
pass
cdef class DelegatedSingleton(Singleton):
__IS_DELEGATED__ = True
cdef class ThreadSafeSingleton(Singleton):
pass
cdef class DelegatedThreadSafeSingleton(ThreadSafeSingleton):
__IS_DELEGATED__ = True
cdef class ThreadLocalSingleton(BaseSingleton):
pass
cdef class DelegatedThreadLocalSingleton(ThreadLocalSingleton):
__IS_DELEGATED__ = True

19
src/dependency_injector/providers/static.pxd Normal file
View File

@ -0,0 +1,19 @@
"""Dependency injector static providers.
Powered by Cython.
"""
from .base cimport Provider
cdef class Object(Provider):
cdef object __provides
cpdef object _provide(self, tuple args, dict kwargs)
cdef class Delegate(Object):
pass
cdef class ExternalDependency(Provider):
cdef type __instance_of

160
src/dependency_injector/providers/static.pyx Normal file
View File

@ -0,0 +1,160 @@
"""Dependency injector static providers.
Powered by Cython.
"""
from dependency_injector.errors import Error
from .base cimport Provider
from .utils cimport (
ensure_is_provider,
represent_provider,
CLASS_TYPES,
)
cdef class Object(Provider):
"""Object provider returns provided instance "as is".
.. py:attribute:: provides
Value that have to be provided.
:type: object
"""
def __init__(self, provides):
"""Initializer.
:param provides: Value that have to be provided.
:type provides: object
"""
self.__provides = provides
super(Object, self).__init__()
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.__provides)
def __repr__(self):
"""Return string representation of provider.
:rtype: str
"""
return self.__str__()
cpdef object _provide(self, tuple args, dict kwargs):
"""Return provided instance.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:rtype: object
"""
return self.__provides
cdef class Delegate(Object):
"""Delegate provider returns provider "as is".
.. py:attribute:: provides
Value that have to be provided.
:type: object
"""
def __init__(self, provides):
"""Initializer.
:param provides: Value that have to be provided.
:type provides: object
"""
super(Delegate, self).__init__(ensure_is_provider(provides))
cdef class ExternalDependency(Provider):
""":py:class:`ExternalDependency` provider describes dependency interface.
This provider is used for description of dependency interface. That might
be useful when dependency could be provided in the client's code only,
but it's interface is known. Such situations could happen when required
dependency has non-determenistic list of dependencies itself.
.. code-block:: python
database_provider = ExternalDependency(sqlite3.dbapi2.Connection)
database_provider.override(Factory(sqlite3.connect, ':memory:'))
database = database_provider()
.. py:attribute:: instance_of
Class of required dependency.
:type: type
"""
def __init__(self, type instance_of):
"""Initializer."""
if not isinstance(instance_of, 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.
:param args: Tuple of context positional arguments.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword arguments.
:type kwargs: dict[str, object]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: object
"""
cdef object instance
if self.__overridden_len == 0:
raise Error('Dependency is not defined')
instance = self._call_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 __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.__instance_of)
def __repr__(self):
"""Return string representation of provider.
:rtype: str
"""
return self.__str__()
def provided_by(self, provider):
"""Set external dependency provider.
:param provider: Provider that provides required dependency.
:type provider: :py:class:`Provider`
:rtype: None
"""
return self.override(provider)

11
src/dependency_injector/providers/utils.pxd
View File

@ -3,6 +3,17 @@
Powered by Cython.
"""
from .base cimport Provider
cdef tuple CLASS_TYPES
cdef class OverridingContext(object):
cdef Provider __overridden
cdef Provider __overriding
cpdef bint is_provider(object instance)
cpdef object ensure_is_provider(object instance)
cpdef bint is_delegated(object instance)

47
src/dependency_injector/providers/utils.pyx
View File

@ -10,16 +10,55 @@ import threading
from dependency_injector.errors import Error
from .base cimport Provider
GLOBAL_LOCK = threading.RLock()
"""Global reentrant lock.
:type: :py:class:`threading.RLock`
"""
cdef tuple CLASS_TYPES
if sys.version_info[0] == 3: # pragma: no cover
_CLASS_TYPES = (type,)
CLASS_TYPES = (type,)
else: # pragma: no cover
_CLASS_TYPES = (type, types.ClassType)
CLASS_TYPES = (type, types.ClassType)
cdef class OverridingContext(object):
"""Provider overriding context.
:py:class:`OverridingContext` is used by :py:meth:`Provider.override` for
implemeting ``with`` contexts. When :py:class:`OverridingContext` is
closed, overriding that was created in this context is dropped also.
.. code-block:: python
with provider.override(another_provider):
assert provider.overridden
assert not provider.overridden
"""
def __init__(self, Provider overridden, Provider overriding):
"""Initializer.
:param overridden: Overridden provider.
:type overridden: :py:class:`Provider`
:param overriding: Overriding provider.
:type overriding: :py:class:`Provider`
"""
self.__overridden = overridden
self.__overriding = overriding
def __enter__(self):
"""Do nothing."""
return self.__overriding
def __exit__(self, *_):
"""Exit overriding context."""
self.__overridden.reset_last_overriding()
cpdef bint is_provider(object instance):
@ -30,7 +69,7 @@ cpdef bint is_provider(object instance):
:rtype: bool
"""
return (not isinstance(instance, _CLASS_TYPES) and
return (not isinstance(instance, CLASS_TYPES) and
getattr(instance, '__IS_PROVIDER__', False) is True)
@ -59,7 +98,7 @@ cpdef bint is_delegated(object instance):
:rtype: bool
"""
return (not isinstance(instance, _CLASS_TYPES) and
return (not isinstance(instance, CLASS_TYPES) and
getattr(instance, '__IS_DELEGATED__', False) is True)

57
tests/unit/providers/test_base.py
View File

@ -84,60 +84,3 @@ class ProviderTests(unittest.TestCase):
self.assertEqual(repr(self.provider),
'<dependency_injector.providers.base.'
'Provider() at {0}>'.format(hex(id(self.provider))))
class DelegateTests(unittest.TestCase):
def setUp(self):
self.delegated = providers.Provider()
self.delegate = providers.Delegate(delegated=self.delegated)
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.delegate))
def test_init_with_not_provider(self):
self.assertRaises(errors.Error, providers.Delegate, delegated=object())
def test_call(self):
delegated1 = self.delegate()
delegated2 = self.delegate()
self.assertIs(delegated1, self.delegated)
self.assertIs(delegated2, self.delegated)
def test_repr(self):
self.assertEqual(repr(self.delegate),
'<dependency_injector.providers.base.'
'Delegate({0}) at {1}>'.format(
repr(self.delegated),
hex(id(self.delegate))))
class ExternalDependencyTests(unittest.TestCase):
def setUp(self):
self.provider = providers.ExternalDependency(instance_of=list)
def test_init_with_not_class(self):
self.assertRaises(errors.Error, providers.ExternalDependency, object())
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.provider))
def test_call_overridden(self):
self.provider.provided_by(providers.Factory(list))
self.assertIsInstance(self.provider(), list)
def test_call_overridden_but_not_instance_of(self):
self.provider.provided_by(providers.Factory(dict))
self.assertRaises(errors.Error, self.provider)
def test_call_not_overridden(self):
self.assertRaises(errors.Error, self.provider)
def test_repr(self):
self.assertEqual(repr(self.provider),
'<dependency_injector.providers.base.'
'ExternalDependency({0}) at {1}>'.format(
repr(list),
hex(id(self.provider))))

10
tests/unit/providers/test_injections.py
View File

@ -25,6 +25,11 @@ class PositionalInjectionTests(unittest.TestCase):
self.assertIs(type(obj2), object)
self.assertIsNot(obj1, obj2)
def test_get_original_value(self):
provider = providers.Factory(object)
injection = providers.PositionalInjection(provider)
self.assertIs(injection.get_original_value(), provider)
class NamedInjectionTests(unittest.TestCase):
@ -50,3 +55,8 @@ class NamedInjectionTests(unittest.TestCase):
self.assertIs(type(obj1), object)
self.assertIs(type(obj2), object)
self.assertIsNot(obj1, obj2)
def test_get_original_value(self):
provider = providers.Factory(object)
injection = providers.NamedInjection('name', provider)
self.assertIs(injection.get_original_value(), provider)

64
tests/unit/providers/test_static.py
View File

@ -2,7 +2,10 @@
import unittest2 as unittest
from dependency_injector import providers
from dependency_injector import (
providers,
errors,
)
class ObjectProviderTests(unittest.TestCase):
@ -25,7 +28,64 @@ class ObjectProviderTests(unittest.TestCase):
some_object = object()
provider = providers.Object(some_object)
self.assertEqual(repr(provider),
'<dependency_injector.providers.base.'
'<dependency_injector.providers.static.'
'Object({0}) at {1}>'.format(
repr(some_object),
hex(id(provider))))
class DelegateTests(unittest.TestCase):
def setUp(self):
self.delegated = providers.Provider()
self.delegate = providers.Delegate(delegated=self.delegated)
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.delegate))
def test_init_with_not_provider(self):
self.assertRaises(errors.Error, providers.Delegate, delegated=object())
def test_call(self):
delegated1 = self.delegate()
delegated2 = self.delegate()
self.assertIs(delegated1, self.delegated)
self.assertIs(delegated2, self.delegated)
def test_repr(self):
self.assertEqual(repr(self.delegate),
'<dependency_injector.providers.static.'
'Delegate({0}) at {1}>'.format(
repr(self.delegated),
hex(id(self.delegate))))
class ExternalDependencyTests(unittest.TestCase):
def setUp(self):
self.provider = providers.ExternalDependency(instance_of=list)
def test_init_with_not_class(self):
self.assertRaises(TypeError, providers.ExternalDependency, object())
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.provider))
def test_call_overridden(self):
self.provider.provided_by(providers.Factory(list))
self.assertIsInstance(self.provider(), list)
def test_call_overridden_but_not_instance_of(self):
self.provider.provided_by(providers.Factory(dict))
self.assertRaises(errors.Error, self.provider)
def test_call_not_overridden(self):
self.assertRaises(errors.Error, self.provider)
def test_repr(self):
self.assertEqual(repr(self.provider),
'<dependency_injector.providers.static.'
'ExternalDependency({0}) at {1}>'.format(
repr(list),
hex(id(self.provider))))