ets-labs/python-dependency-injector
1
1
Fork 0
mirror of https://github.com/ets-labs/python-dependency-injector.git synced 2025-02-07 07:00:49 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #143 from ets-labs/138_cython_providers_v3

Browse Source 
Cython providers
This commit is contained in:
Roman 2016-11-11 17:27:54 +02:00 committed by GitHub
commit 8259de6dd7
38 changed files with 2753 additions and 1541 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
README.rst
View File

@ -16,6 +16,9 @@ dependency injection pattern in formal, pretty, Pythonic way.
+ Documentation.
+ Semantic versioning.
*Dependency Injector* providers are implemented as C extension types using
Cython.
Status
------

1
docs/api/index.rst
View File

@ -7,5 +7,4 @@ API Documentation
top_level
providers
containers
utils
errors

4
docs/api/providers.rst
View File

@ -1,10 +1,6 @@
``dependency_injector.providers``
---------------------------------
.. image:: /images/providers/providers_class_diagram.png
:width: 100%
:align: center
.. automodule:: dependency_injector.providers
:members:
:inherited-members:

2
docs/containers/index.rst
View File

@ -17,7 +17,7 @@ Also, for both of these and some other cases, it might be useful to attach
some init / shutdown functionality or something else, that deals with group
of providers.
Containers module API docs - :py:mod:`dependency_injector.containers`.
Containers package API docs - :py:mod:`dependency_injector.containers`.
.. toctree::
:maxdepth: 2

BIN
docs/images/providers/providers_class_diagram.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 139 KiB

15
docs/index.rst
View File

@ -8,22 +8,33 @@ Dependency Injector --- Dependency injection microframework for Python
:description: Dependency Injector is a dependency injection microframework
for Python. It was designed to be unified, developer-friendly
tool that helps to implement dependency injection pattern in
formal, pretty, Pythonic way.
formal, pretty, Pythonic way. Dependency Injector framework
key features are: Easy, smart, pythonic style; Obvious, clear
structure; Extensibility and flexibility; High performance;
Memory efficiency; Thread safety; Documentation; Semantic
versioning; Dependency Injector providers are implemented as
C extension types using Cython.
.. _index:
*Dependency Injector* is a dependency injection microframework for Python.
It was designed to be unified, developer-friendly tool that helps to implement
dependency injection pattern in formal, pretty, Pythonic way.
dependency injection pattern in formal, pretty, Pythonic way.
*Dependency Injector* framework key features are:
+ Easy, smart, pythonic style.
+ Obvious, clear structure.
+ Extensibility and flexibility.
+ High performance.
+ Memory efficiency.
+ Thread safety.
+ Documentation.
+ Semantic versioning.
*Dependency Injector* providers are implemented as C extension types using
Cython.
Status
------

2
docs/introduction/di_in_python.rst
View File

@ -121,7 +121,7 @@ code for specification of dependencies. Nevertheless, this disadvantage could
be easily avoided by creating inversion of control container (IoC container).
Example of creation of several inversion of control containers (IoC containers)
using *Dependency Injector*:
using :doc:`Dependency Injector <../index>`:
.. literalinclude:: ../../examples/miniapps/engines_cars/example_ioc_containers.py
:language: python

42
docs/main/changelog.rst
View File

@ -9,16 +9,40 @@ follows `Semantic versioning`_
Development version
-------------------
- Add ``dependency_injector.injections`` module (C extension).
- Transfer ``dependency_injector.utils`` module to Cython (C extension).
- Transfer ``dependency_injector.errors`` module to Cython (C extension).
- Remove ``@inject`` decorator.
- Add makefile (``clean``, ``test``, ``build``, ``install``, ``uninstall``
& ``publish`` commands).
- Update repository structure:
- **Providers**
- Sources are moved under ``src``.
- Tests are moved under ``tests/unit``.
1. All providers from ``dependency_injector.providers`` package are
implemented as C extension types using Cython.
2. Add ``BaseSingleton`` super class for all singleton providers.
3. Make ``Singleton`` provider not thread-safe. It makes performance of
``Singleton`` provider 10x times faster.
4. Add ``ThreadSafeSingleton`` provider - thread-safe version of
``Singleton`` provider.
5. Add ``ThreadLocalSingleton`` provider - ``Singleton`` provider that uses
thread-local storage.
6. Remove ``provides`` attribute from ``Factory`` and ``Singleton``
providers.
7. Add ``set_args()`` and ``clear_args()`` methods for ``Callable``,
``Factory`` and ``Singleton`` providers.
- **Containers**
1. Module ``dependency_injector.containers`` was splitted into submodules
without any functional changes.
- **Utils**
1. Module ``dependency_injector.utils`` is splitted into
``dependency_injector.containers`` and ``dependency_injector.providers``.
- **Miscellaneous**
1. Remove ``@inject`` decorator.
2. Add makefile (``clean``, ``test``, ``build``, ``install``, ``uninstall``
& ``publish`` commands).
3. Update repository structure:
1. Sources are moved under ``src/`` folder.
2. Tests are moved under ``tests/unit/`` folder.
.. - No features.

9
docs/main/installation.rst
View File

@ -12,8 +12,13 @@ framework can be installed from PyPi_:
pip install dependency_injector
# Installing particular version:
pip install dependency_injector==2.0.0
pip install dependency-injector==3.0.0
.. note::
Some components of *Dependency Injector* are implemented as C extension types.
*Dependency Injector* is distributed as an archive with a source code, so
C compiler and Python header files are required for the installation.
Sources can be cloned from GitHub_:
.. code-block:: bash
@ -30,7 +35,7 @@ Verification of currently installed version could be done using
>>> from dependency_injector import VERSION
>>> VERSION
'2.0.0'
'3.0.0'
.. _PyPi: https://pypi.python.org/pypi/dependency_injector
.. _GitHub: https://github.com/ets-labs/python-dependency-injector

11
docs/providers/index.rst
View File

@ -12,16 +12,7 @@ Current documentation section consists from description of standard providers
library and some useful information like overriding of providers and writing
of custom providers.
All providers are validated in multithreading environment and considered to
be thread safe.
+ Base providers class is: :py:class:`dependency_injector.providers.Provider`
+ Providers module API docs: :py:mod:`dependency_injector.providers`
+ Providers class diagram:
.. image:: /images/providers/providers_class_diagram.png
:width: 100%
:align: center
Providers package API docs - :py:mod:`dependency_injector.providers`
.. toctree::
:maxdepth: 2

20
docs/providers/singleton.rst
View File

@ -21,7 +21,7 @@ Singleton providers resetting
Created and memorized by :py:class:`Singleton` instance can be reset. Reset of
:py:class:`Singleton`'s memorized instance is done by clearing reference to
it. Further lifecycle of memorized instance is out of :py:class:`Singleton`
it. Further lifecycle of memorized instance is out of :py:class:`Singleton`
provider's control and dependes on garbage collection strategy.
Example:
@ -33,15 +33,15 @@ Example:
Singleton providers and injections
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
:py:class:`Singleton` provider extends :py:class:`Factory` provider, so, all
of the rules about injections are the same, as for :py:class:`Factory`
provider.
:py:class:`Singleton` provider has same interface as :py:class:`Factory`
provider, so, all of the rules about injections are the same, as for
:py:class:`Factory` provider.
.. note::
Due that :py:class:`Singleton` provider creates specified class instance
only on the first call, all injections are done once, during the first
call, also. Every next call, while instance has been already created
call. Every next call, while instance has been already created
and memorized, no injections are done, :py:class:`Singleton` provider just
returns memorized earlier instance.
@ -81,10 +81,12 @@ Specialization of :py:class:`Singleton` providers is the same as
Singleton providers and multi-threading
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
:py:class:`Singleton` provider is thread-safe and could be used in
multi-threading applications without any negative impact. Race condition on
singleton's initialization is escaped by using a global reentrant mutex -
:py:obj:`dependency_injector.utils.GLOBAL_LOCK`.
:py:class:`Singleton` provider is NOT thread-safe and should be used in
multi-threading applications with manually controlled locking.
:py:class:`ThreadSafeSingleton` is a thread-safe version of
:py:class:`Singleton` and could be used in multi-threading applications
without any additional locking.
Also there could be a need to use thread-scoped singletons and there is a
special provider for such case - :py:class:`ThreadLocalSingleton`.

16
setup.py
View File

@ -47,6 +47,22 @@ 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.injections',
['src/dependency_injector/providers/injections.c'],
define_macros=defined_macros,

15
src/dependency_injector/containers/utils.py
View File

@ -1,20 +1,12 @@
"""Dependency injector container utils."""
import copy as _copy
import types
import six
from dependency_injector.providers import deepcopy
from dependency_injector.errors import Error
if six.PY2: # pragma: no cover
_copy._deepcopy_dispatch[types.MethodType] = \
lambda obj, memo: type(obj)(obj.im_func,
_copy.deepcopy(obj.im_self, memo),
obj.im_class)
def is_container(instance):
"""Check if instance is container instance.
@ -74,11 +66,6 @@ def copy(container):
return _decorator
def deepcopy(instance, memo=None):
"""Make full copy of instance."""
return _copy.deepcopy(instance, memo)
def _check_provider_type(cls, provider):
if not isinstance(provider, cls.provider_type):
raise Error('{0} can contain only {1} '

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

@ -2,46 +2,51 @@
from .base import (
Provider,
Delegate,
Object,
Delegate,
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 .injections import (
Injection,
PositionalInjection,
NamedInjection,
)
from .utils import (
is_provider,
ensure_is_provider,
is_delegated,
represent_provider,
deepcopy,
)
__all__ = (
'Provider',
'Delegate',
'Object',
'Delegate',
'ExternalDependency',
'OverridingContext',
'override',
'Callable',
'DelegatedCallable',
@ -49,19 +54,24 @@ __all__ = (
'Factory',
'DelegatedFactory',
'BaseSingleton',
'Singleton',
'DelegatedSingleton',
'ThreadSafeSingleton',
'DelegatedThreadSafeSingleton',
'ThreadLocalSingleton',
'DelegatedThreadLocalSingleton',
'GLOBAL_LOCK',
'Injection',
'PositionalInjection',
'NamedInjection',
'is_provider',
'ensure_is_provider',
'is_delegated',
'represent_provider',
'Injection',
'PositionalInjection',
'NamedInjection',
'deepcopy',
)

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

@ -0,0 +1,31 @@
"""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)
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
cdef class OverridingContext(object):
cdef Provider __overridden
cdef Provider __overriding

348
src/dependency_injector/providers/base.py → src/dependency_injector/providers/base.pyx
View File

@ -1,17 +1,22 @@
"""Dependency injector base providers."""
"""Dependency injector base providers.
import six
Powered by Cython.
"""
cimport cython
from dependency_injector.errors import Error
from .utils import (
from .utils cimport (
CLASS_TYPES,
is_provider,
ensure_is_provider,
represent_provider,
deepcopy,
)
@six.python_2_unicode_compatible
class Provider(object):
cdef class Provider(object):
"""Base provider class.
:py:class:`Provider` is callable (implements ``__call__`` method). Every
@ -59,38 +64,53 @@ class Provider(object):
"""
__IS_PROVIDER__ = True
__OPTIMIZED_CALLS__ = True
__slots__ = ('overridden', 'provide', '__call__')
def __init__(self):
"""Initializer."""
self.overridden = tuple()
self.__overridden = tuple()
self.__overridden_len = 0
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.
def __call__(self, *args, **kwargs):
"""Return provided object.
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.
Callable interface implementation.
"""
raise NotImplementedError()
if self.__overridden_len != 0:
return self._call_last_overriding(args, kwargs)
return self._provide(args, kwargs)
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 __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
def provide_injection(self):
"""Injection strategy implementation.
copied = self.__class__()
:rtype: object
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return self.provide()
return represent_provider(provider=self, provides=None)
def __repr__(self):
"""Return string representation of provider.
:rtype: str
"""
return self.__str__()
@property
def overridden(self):
"""Return tuple of overriding providers."""
return self.__overridden
def override(self, provider):
"""Override provider with another provider.
@ -100,8 +120,8 @@ class Provider(object):
:raise: :py:exc:`dependency_injector.errors.Error`
:return: Overriding provider.
:rtype: :py:class:`Provider`
:return: Overriding context.
:rtype: :py:class:`OverridingContext`
"""
if provider is self:
raise Error('Provider {0} could not be overridden '
@ -110,14 +130,13 @@ class Provider(object):
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
self.__overridden += (provider,)
self.__overridden_len += 1
return OverridingContext(self, provider)
@cython.boundscheck(False)
@cython.wraparound(False)
def reset_last_overriding(self):
"""Reset last overriding provider.
@ -126,26 +145,19 @@ class Provider(object):
:rtype: None
"""
if not self.overridden:
if self.__overridden_len == 0:
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
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()
# Enable __call__() / _provide() optimization
if self.__class__.__OPTIMIZED_CALLS__:
self.__call__ = self.provide = self._provide
self.__overridden = tuple()
self.__overridden_len = 0
def delegate(self):
"""Return provider's delegate.
@ -154,73 +166,25 @@ class Provider(object):
"""
return Delegate(self)
def __str__(self):
"""Return string representation of provider.
cpdef object _provide(self, tuple args, dict kwargs):
"""Providing strategy implementation.
:rtype: str
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.
"""
return represent_provider(provider=self, provides=None)
raise NotImplementedError()
__repr__ = __str__
@cython.boundscheck(False)
@cython.wraparound(False)
cpdef object _call_last_overriding(self, tuple args, dict kwargs):
"""Call last overriding provider and return result."""
return <object>self.__overridden[self.__overridden_len - 1](*args,
**kwargs)
@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".
cdef class Object(Provider):
"""Object provider returns provided instance "as is".
.. py:attribute:: provides
@ -229,18 +193,43 @@ class Object(Provider):
:type: object
"""
__slots__ = ('provides',)
def __init__(self, provides):
"""Initializer.
:param provides: Value that have to be provided.
:type provides: object
"""
self.provides = provides
self.__provides = provides
super(Object, self).__init__()
def _provide(self, *args, **kwargs):
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = self.__class__(deepcopy(self.__provides, memo))
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
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.
@ -251,20 +240,29 @@ class Object(Provider):
:rtype: object
"""
return self.provides
return self.__provides
def __str__(self):
"""Return string representation of provider.
:rtype: str
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
"""
return represent_provider(provider=self, provides=self.provides)
__repr__ = __str__
super(Delegate, self).__init__(ensure_is_provider(provides))
@six.python_2_unicode_compatible
class ExternalDependency(Provider):
cdef class ExternalDependency(Provider):
""":py:class:`ExternalDependency` provider describes dependency interface.
This provider is used for description of dependency interface. That might
@ -286,18 +284,24 @@ class ExternalDependency(Provider):
:type: type
"""
__OPTIMIZED_CALLS__ = False
__slots__ = ('instance_of',)
def __init__(self, instance_of):
def __init__(self, type 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__
self.__instance_of = instance_of
super(ExternalDependency, self).__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = self.__class__(self.__instance_of)
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
def __call__(self, *args, **kwargs):
"""Return provided instance.
@ -311,10 +315,12 @@ class ExternalDependency(Provider):
:rtype: object
"""
if not self.overridden:
cdef object instance
if self.__overridden_len == 0:
raise Error('Dependency is not defined')
instance = self._call_last_overriding(*args, **kwargs)
instance = self._call_last_overriding(args, kwargs)
if not isinstance(instance, self.instance_of):
raise Error('{0} is not an '.format(instance) +
@ -322,6 +328,25 @@ class ExternalDependency(Provider):
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__()
@property
def instance_of(self):
"""Return class of required dependency."""
return self.__instance_of
def provided_by(self, provider):
"""Set external dependency provider.
@ -332,17 +357,8 @@ class ExternalDependency(Provider):
"""
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):
cdef class OverridingContext(object):
"""Provider overriding context.
:py:class:`OverridingContext` is used by :py:meth:`Provider.override` for
@ -356,7 +372,7 @@ class OverridingContext(object):
assert not provider.overridden
"""
def __init__(self, overridden, overriding):
def __init__(self, Provider overridden, Provider overriding):
"""Initializer.
:param overridden: Overridden provider.
@ -365,52 +381,14 @@ class OverridingContext(object):
:param overriding: Overriding provider.
:type overriding: :py:class:`Provider`
"""
self.overridden = overridden
self.overriding = overriding
self.__overridden = overridden
self.__overriding = overriding
super(OverridingContext, self).__init__()
def __enter__(self):
"""Do nothing."""
return self.overriding
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))
self.__overridden.reset_last_overriding()

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

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

@ -0,0 +1,42 @@
"""Dependency injector callable providers.
Powered by Cython.
"""
from .base cimport Provider
from .injections cimport (
PositionalInjection,
NamedInjection,
__provide_positional_args,
__provide_keyword_args,
)
cdef class Callable(Provider):
cdef object __provides
cdef tuple __args
cdef int __args_len
cdef tuple __kwargs
cdef int __kwargs_len
cpdef object _provide(self, tuple args, dict kwargs)
cdef inline object __provide(self, tuple args, dict kwargs):
cdef tuple positional_args
cdef dict keyword_args
positional_args = __provide_positional_args(args,
self.__args,
self.__args_len)
keyword_args = __provide_keyword_args(kwargs,
self.__kwargs,
self.__kwargs_len)
return self.__provides(*positional_args, **keyword_args)
cdef class DelegatedCallable(Callable):
pass

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

@ -0,0 +1,208 @@
"""Dependency injector callable providers.
Powered by Cython.
"""
from dependency_injector.errors import Error
from .base cimport Provider
from .injections cimport (
PositionalInjection,
NamedInjection,
parse_positional_injections,
parse_named_injections,
)
from .utils cimport (
represent_provider,
deepcopy,
)
cdef class Callable(Provider):
r"""Callable provider calls wrapped callable on every call.
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)
"""
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Wrapped callable.
:type provides: callable
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
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.__args_len = 0
self.set_args(*args)
self.__kwargs = tuple()
self.__kwargs_len = 0
self.set_kwargs(**kwargs)
super(Callable, self).__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = self.__class__(self.provides,
*deepcopy(self.args, memo),
**deepcopy(self.kwargs, memo))
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self, provides=self.__provides)
@property
def provides(self):
"""Return wrapped callable."""
return self.__provides
@property
def args(self):
"""Return positional argument injections."""
cdef int index
cdef PositionalInjection arg
cdef list args
args = list()
for index in range(self.__args_len):
arg = self.__args[index]
args.append(arg.__value)
return tuple(args)
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)
self.__args_len = len(self.__args)
return self
def set_args(self, *args):
"""Set postional argument injections.
Existing positional argument injections are dropped.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__args = parse_positional_injections(args)
self.__args_len = len(self.__args)
return self
def clear_args(self):
"""Drop postional argument injections.
:return: Reference ``self``
"""
self.__args = tuple()
self.__args_len = len(self.__args)
return self
@property
def kwargs(self):
"""Return keyword argument injections."""
cdef int index
cdef NamedInjection kwarg
cdef dict kwargs
kwargs = dict()
for index in range(self.__kwargs_len):
kwarg = self.__kwargs[index]
kwargs[kwarg.__name] = kwarg.__value
return kwargs
def add_kwargs(self, **kwargs):
"""Add keyword argument injections.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__kwargs += parse_named_injections(kwargs)
self.__kwargs_len = len(self.__kwargs)
return self
def set_kwargs(self, **kwargs):
"""Set keyword argument injections.
Existing keyword argument injections are dropped.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__kwargs = parse_named_injections(kwargs)
self.__kwargs_len = len(self.__kwargs)
return self
def clear_kwargs(self):
"""Drop keyword argument injections.
:return: Reference ``self``
"""
self.__kwargs = tuple()
self.__kwargs_len = len(self.__kwargs)
return self
cpdef object _provide(self, tuple args, dict kwargs):
"""Return result of provided callable's call."""
return self.__provide(args, kwargs)
cdef class DelegatedCallable(Callable):
"""Callable that is injected "as is".
DelegatedCallable is a :py:class:`Callable`, that is injected "as is".
"""
__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

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

@ -0,0 +1,33 @@
"""Dependency injector factory providers.
Powered by Cython.
"""
from .base cimport Provider
from .callables cimport Callable
from .injections cimport __inject_attributes
cdef class Factory(Provider):
cdef Callable __instantiator
cdef tuple __attributes
cdef int __attributes_len
cpdef object _provide(self, tuple args, dict kwargs)
cdef inline object __provide(self, tuple args, dict kwargs):
cdef object instance
instance = self.__instantiator.__provide(args, kwargs)
if self.__attributes_len > 0:
__inject_attributes(instance,
self.__attributes,
self.__attributes_len)
return instance
cdef class DelegatedFactory(Factory):
pass

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

@ -0,0 +1,270 @@
"""Dependency injector factory providers.
Powered by Cython.
"""
from dependency_injector.errors import Error
from .base cimport Provider
from .callables cimport Callable
from .injections cimport (
NamedInjection,
parse_named_injections,
)
from .utils cimport (
represent_provider,
deepcopy,
)
cdef class Factory(Provider):
r"""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
"""
provided_type = None
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Provided type.
:type provides: type
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
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.__instantiator = Callable(provides, *args, **kwargs)
self.__attributes = tuple()
self.__attributes_len = 0
super(Factory, self).__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = self.__class__(self.cls,
*deepcopy(self.args, memo),
**deepcopy(self.kwargs, memo))
copied.set_attributes(**deepcopy(self.attributes, memo))
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self,
provides=self.__instantiator.provides)
@property
def cls(self):
"""Return provided type."""
return self.__instantiator.provides
@property
def args(self):
"""Return positional argument injections."""
return self.__instantiator.args
def add_args(self, *args):
"""Add __init__ postional argument injections.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.add_args(*args)
return self
def set_args(self, *args):
"""Set __init__ postional argument injections.
Existing __init__ positional argument injections are dropped.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.set_args(*args)
return self
def clear_args(self):
"""Drop __init__ postional argument injections.
:return: Reference ``self``
"""
self.__instantiator.clear_args()
return self
@property
def kwargs(self):
"""Return keyword argument injections."""
return self.__instantiator.kwargs
def add_kwargs(self, **kwargs):
"""Add __init__ keyword argument injections.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__instantiator.add_kwargs(**kwargs)
return self
def set_kwargs(self, **kwargs):
"""Set __init__ keyword argument injections.
Existing __init__ keyword argument injections are dropped.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__instantiator.set_kwargs(**kwargs)
return self
def clear_kwargs(self):
"""Drop __init__ keyword argument injections.
:return: Reference ``self``
"""
self.__instantiator.clear_kwargs()
return self
@property
def attributes(self):
"""Return attribute injections."""
cdef int index
cdef NamedInjection attribute
cdef dict attributes
attributes = dict()
for index in range(self.__attributes_len):
attribute = self.__attributes[index]
attributes[attribute.__name] = attribute.__value
return attributes
def add_attributes(self, **kwargs):
"""Add attribute injections.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__attributes += parse_named_injections(kwargs)
self.__attributes_len = len(self.__attributes)
return self
def set_attributes(self, **kwargs):
"""Set attribute injections.
Existing attribute injections are dropped.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__attributes = parse_named_injections(kwargs)
self.__attributes_len = len(self.__attributes)
return self
def clear_attributes(self):
"""Drop attribute injections.
:return: Reference ``self``
"""
self.__attributes = tuple()
self.__attributes_len = len(self.__attributes)
return self
cpdef object _provide(self, tuple args, dict kwargs):
"""Return new instance."""
return self.__provide(args, kwargs)
cdef class DelegatedFactory(Factory):
"""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
"""
__IS_DELEGATED__ = True

49
src/dependency_injector/providers/injections.pxd
View File

@ -6,7 +6,7 @@ Powered by Cython.
cimport cython
cdef class Injection:
cdef class Injection(object):
pass
@ -40,9 +40,9 @@ cdef class NamedInjection(Injection):
@cython.boundscheck(False)
@cython.wraparound(False)
cdef inline tuple __provide_positional_args(tuple inj_args,
int inj_args_len,
tuple args):
cdef inline tuple __provide_positional_args(tuple args,
tuple inj_args,
int inj_args_len):
cdef int index
cdef list positional_args
cdef PositionalInjection injection
@ -53,29 +53,48 @@ cdef inline tuple __provide_positional_args(tuple inj_args,
positional_args = list()
for index in range(inj_args_len):
injection = <PositionalInjection>inj_args[index]
positional_args.append(injection.get_value())
positional_args.append(injection.__get_value())
positional_args.extend(args)
return positional_args
return tuple(positional_args)
@cython.boundscheck(False)
@cython.wraparound(False)
cdef inline dict __provide_keyword_args(tuple inj_kwargs,
int inj_kwargs_len,
dict kwargs):
cdef inline dict __provide_keyword_args(dict kwargs,
tuple inj_kwargs,
int inj_kwargs_len):
cdef int index
cdef object name
cdef NamedInjection kw_injection
if inj_kwargs_len == 0:
return kwargs
for index in range(inj_kwargs_len):
kw_injection = <NamedInjection>inj_kwargs[index]
kwargs[kw_injection.get_name()] = kw_injection.get_value()
if len(kwargs) == 0:
for index in range(inj_kwargs_len):
kw_injection = <NamedInjection>inj_kwargs[index]
name = kw_injection.__get_name()
kwargs[name] = kw_injection.__get_value()
else:
for index in range(inj_kwargs_len):
kw_injection = <NamedInjection>inj_kwargs[index]
name = kw_injection.__get_name()
if name not in kwargs:
kwargs[name] = kw_injection.__get_value()
return kwargs
@cython.boundscheck(False)
@cython.wraparound(False)
cdef inline object __inject_attributes(object instance,
tuple attributes,
int attributes_len):
cdef NamedInjection attr_injection
for index in range(attributes_len):
attr_injection = <NamedInjection>attributes[index]
setattr(instance,
attr_injection.__get_name(),
attr_injection.__get_value())
cpdef tuple parse_positional_injections(tuple args)
cpdef tuple parse_named_injections(dict kwargs)

26
src/dependency_injector/providers/injections.pyx
View File

@ -8,10 +8,11 @@ cimport cython
from .utils cimport (
is_provider,
is_delegated,
deepcopy,
)
cdef class Injection:
cdef class Injection(object):
"""Abstract injection class."""
@ -23,7 +24,16 @@ cdef class PositionalInjection(Injection):
self.__value = value
self.__is_provider = <int>is_provider(value)
self.__is_delegated = <int>is_delegated(value)
self.__call = <int>self.__is_provider == 1 and self.__is_delegated == 0
self.__call = <int>(self.__is_provider == 1 and
self.__is_delegated == 0)
super(PositionalInjection, self).__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
return self.__class__(deepcopy(self.__value, memo))
def get_value(self):
"""Return injection value."""
@ -43,7 +53,17 @@ cdef class NamedInjection(Injection):
self.__value = value
self.__is_provider = <int>is_provider(value)
self.__is_delegated = <int>is_delegated(value)
self.__call = <int>self.__is_provider == 1 and self.__is_delegated == 0
self.__call = <int>(self.__is_provider == 1 and
self.__is_delegated == 0)
super(NamedInjection, self).__init__()
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
return self.__class__(deepcopy(self.__name, memo),
deepcopy(self.__value, memo))
def get_name(self):
"""Return injection value."""

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

@ -0,0 +1,64 @@
"""Dependency injector singleton providers.
Powered by Cython.
"""
from .base cimport Provider
from .factories cimport Factory
cdef class BaseSingleton(Provider):
cdef Factory __instantiator
cdef class Singleton(BaseSingleton):
cdef object __storage
cpdef object _provide(self, tuple args, dict kwargs)
cdef inline object __provide(self, tuple args, dict kwargs):
if self.__storage is None:
self.__storage = self.__instantiator.__provide(args, kwargs)
return self.__storage
cdef class DelegatedSingleton(Singleton):
pass
cdef class ThreadSafeSingleton(BaseSingleton):
cdef object __storage
cdef object __lock
cpdef object _provide(self, tuple args, dict kwargs)
cdef inline object __provide(self, tuple args, dict kwargs):
with self.__lock:
if self.__storage is None:
self.__storage = self.__instantiator.__provide(args, kwargs)
return self.__storage
cdef class DelegatedThreadSafeSingleton(ThreadSafeSingleton):
pass
cdef class ThreadLocalSingleton(BaseSingleton):
cdef object __storage
cpdef object _provide(self, tuple args, dict kwargs)
cdef inline object __provide(self, tuple args, dict kwargs):
cdef object instance
try:
instance = self.__storage.instance
except AttributeError:
instance = self.__instantiator.__provide(args, kwargs)
self.__storage.instance = instance
finally:
return instance
cdef class DelegatedThreadLocalSingleton(ThreadLocalSingleton):
pass

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

@ -0,0 +1,393 @@
"""Dependency injector singleton providers.
Powered by Cython.
"""
import threading
from dependency_injector.errors import Error
from .base cimport Provider
from .factories cimport Factory
from .utils cimport (
represent_provider,
deepcopy,
)
GLOBAL_LOCK = threading.RLock()
"""Global reentrant lock.
:type: :py:class:`threading.RLock`
"""
cdef class BaseSingleton(Provider):
"""Base class of singleton providers."""
provided_type = None
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Provided type.
:type provides: type
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
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.__instantiator = Factory(provides, *args, **kwargs)
super(BaseSingleton, self).__init__()
def __str__(self):
"""Return string representation of provider.
:rtype: str
"""
return represent_provider(provider=self,
provides=self.__instantiator.cls)
def __deepcopy__(self, memo):
"""Create and return full copy of provider."""
copied = memo.get(id(self))
if copied is not None:
return copied
copied = self.__class__(self.cls,
*deepcopy(self.args, memo),
**deepcopy(self.kwargs, memo))
copied.set_attributes(**deepcopy(self.attributes, memo))
for overriding_provider in self.overridden:
copied.override(deepcopy(overriding_provider, memo))
return copied
@property
def cls(self):
"""Return provided type."""
return self.__instantiator.cls
@property
def args(self):
"""Return positional argument injections."""
return self.__instantiator.args
def add_args(self, *args):
"""Add __init__ postional argument injections.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.add_args(*args)
return self
def set_args(self, *args):
"""Set __init__ postional argument injections.
Existing __init__ positional argument injections are dropped.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.set_args(*args)
return self
def clear_args(self):
"""Drop __init__ postional argument injections.
:return: Reference ``self``
"""
self.__instantiator.clear_args()
return self
@property
def kwargs(self):
"""Return keyword argument injections."""
return self.__instantiator.kwargs
def add_kwargs(self, **kwargs):
"""Add __init__ keyword argument injections.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__instantiator.add_kwargs(**kwargs)
return self
def set_kwargs(self, **kwargs):
"""Set __init__ keyword argument injections.
Existing __init__ keyword argument injections are dropped.
:param kwargs: Dictionary of injections.
:type kwargs: dict
:return: Reference ``self``
"""
self.__instantiator.set_kwargs(**kwargs)
return self
def clear_kwargs(self):
"""Drop __init__ keyword argument injections.
:return: Reference ``self``
"""
self.__instantiator.clear_kwargs()
return self
@property
def attributes(self):
"""Return attribute injections."""
return self.__instantiator.attributes
def add_attributes(self, **kwargs):
"""Add attribute injections.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.add_attributes(**kwargs)
return self
def set_attributes(self, **kwargs):
"""Set attribute injections.
Existing attribute injections are dropped.
:param args: Tuple of injections.
:type args: tuple
:return: Reference ``self``
"""
self.__instantiator.set_attributes(**kwargs)
return self
def clear_attributes(self):
"""Drop attribute injections.
:return: Reference ``self``
"""
self.__instantiator.clear_attributes()
return self
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
raise NotImplementedError()
cdef class Singleton(BaseSingleton):
"""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.
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
"""
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Provided type.
:type provides: type
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
self.__storage = None
super(Singleton, self).__init__(provides, *args, **kwargs)
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
self.__storage = None
cpdef object _provide(self, tuple args, dict kwargs):
"""Return single instance."""
return self.__provide(args, kwargs)
cdef class DelegatedSingleton(Singleton):
"""Delegated singleton 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
"""
__IS_DELEGATED__ = True
cdef class ThreadSafeSingleton(BaseSingleton):
"""Thread-safe singleton provider."""
def __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Provided type.
:type provides: type
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
self.__storage = None
self.__lock = GLOBAL_LOCK
super(ThreadSafeSingleton, self).__init__(provides, *args, **kwargs)
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
self.__storage = None
cpdef object _provide(self, tuple args, dict kwargs):
"""Return single instance."""
return self.__provide(args, kwargs)
cdef class DelegatedThreadSafeSingleton(ThreadSafeSingleton):
"""Delegated thread-safe singleton 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
"""
__IS_DELEGATED__ = True
cdef class ThreadLocalSingleton(BaseSingleton):
"""Thread-local singleton provides single objects in scope of thread.
.. 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 __init__(self, provides, *args, **kwargs):
"""Initializer.
:param provides: Provided type.
:type provides: type
:param args: Tuple of positional argument injections.
:type args: tuple[object]
:param kwargs: Dictionary of context keyword argument injections.
:type kwargs: dict[str, object]
"""
self.__storage = threading.local()
super(ThreadLocalSingleton, self).__init__(provides, *args, **kwargs)
def reset(self):
"""Reset cached instance, if any.
:rtype: None
"""
self.__storage.instance = None
cpdef object _provide(self, tuple args, dict kwargs):
"""Return single instance."""
return self.__provide(args, kwargs)
cdef class DelegatedThreadLocalSingleton(ThreadLocalSingleton):
"""Delegated thread-local singleton 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
"""
__IS_DELEGATED__ = True

31
src/dependency_injector/providers/utils.h Normal file
View File

@ -0,0 +1,31 @@
/* Generated by Cython 0.25.1 */
#ifndef __PYX_HAVE__dependency_injector__providers__utils
#define __PYX_HAVE__dependency_injector__providers__utils
#ifndef __PYX_HAVE_API__dependency_injector__providers__utils
#ifndef __PYX_EXTERN_C
#ifdef __cplusplus
#define __PYX_EXTERN_C extern "C"
#else
#define __PYX_EXTERN_C extern
#endif
#endif
#ifndef DL_IMPORT
#define DL_IMPORT(_T) _T
#endif
__PYX_EXTERN_C DL_IMPORT(PyObject) *__pyx_v_19dependency_injector_9providers_5utils_CLASS_TYPES;
#endif /* !__PYX_HAVE_API__dependency_injector__providers__utils */
#if PY_MAJOR_VERSION < 3
PyMODINIT_FUNC initutils(void);
#else
PyMODINIT_FUNC PyInit_utils(void);
#endif
#endif /* !__PYX_HAVE__dependency_injector__providers__utils */

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

@ -3,7 +3,11 @@
Powered by Cython.
"""
cdef public object CLASS_TYPES
cpdef bint is_provider(object instance)
cpdef object ensure_is_provider(object instance)
cpdef bint is_delegated(object instance)
cpdef str represent_provider(object provider, object provides)
cpdef object deepcopy(object instance, dict memo=*)

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

@ -3,23 +3,23 @@
Powered by Cython.
"""
import copy
import sys
import types
import threading
from dependency_injector.errors import Error
GLOBAL_LOCK = threading.RLock()
"""Global reentrant lock.
:type: :py:class:`threading.RLock`
"""
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)
copy._deepcopy_dispatch[types.MethodType] = \
lambda obj, memo: type(obj)(obj.im_func,
copy.deepcopy(obj.im_self, memo),
obj.im_class)
cpdef bint is_provider(object instance):
@ -30,7 +30,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 +59,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)
@ -80,3 +80,7 @@ cpdef str represent_provider(object provider, object provides):
provider.__class__.__name__)),
provides=repr(provides) if provides is not None else '',
address=hex(id(provider)))
cpdef object deepcopy(object instance, dict memo=None):
"""Return full copy of provider or container with providers."""
return copy.deepcopy(instance, memo)

177
tests/performance/test.py Normal file
View File

@ -0,0 +1,177 @@
"""Test providers performance."""
import time
import gc
import dependency_injector.providers
class Tester(object):
"""Performance tester for provider module implementations."""
def __init__(self, provider_modules, duration_factor):
"""Initializer."""
self.provider_modules = provider_modules
self.tests = [getattr(self, name)
for name in dir(self)
if name.startswith('test')]
self.total_time = 0
self.duration_factor = duration_factor
def run(self):
"""Run all tests for all provider modules."""
for module in self.provider_modules:
print('\n')
print('Running tests for module - "{module}":'
.format(module=module.__name__))
gc.disable()
for test in self.tests:
start_time = time.time()
test(module)
self.total_time = time.time() - start_time
print('Test "{test}" took - {time}'
.format(test=test.__name__,
time=self.total_time))
gc.collect()
gc.enable()
print('\n')
# def test_simple_object(self, providers):
# """Test simple object's creation."""
# class Test(object):
# pass
#
# for x in xrange(int(5000000 * self.duration_factor)):
# Test()
#
# def test_simple_object_factory(self, providers):
# """Test simple object's factory."""
# class Test(object):
# pass
#
# test_factory = providers.Factory(Test)
# for x in xrange(int(5000000 * self.duration_factor)):
# test_factory()
#
# def test_3_ctx_positional_injections(self, providers):
# """Test factory with 3 context positional injections."""
# class Test(object):
# def __init__(self, a, b, c):
# pass
#
# for x in xrange(int(5000000 * self.duration_factor)):
# Test(1, 2, 3)
#
# def test_factory_3_ctx_positional_injections(self, providers):
# """Test factory with 3 context positional injections."""
# class Test(object):
# def __init__(self, a, b, c):
# pass
#
# test_factory = providers.Factory(Test)
# for x in xrange(int(5000000 * self.duration_factor)):
# test_factory(1, 2, 3)
def test_3_kw_injections(self, providers):
"""Test 3 keyword argument injections."""
class A(object):
pass
class B(object):
pass
class C(object):
pass
class Test(object):
def __init__(self, a, b, c):
pass
for x in xrange(int(5000000 * self.duration_factor)):
Test(a=A(), b=B(), c=C())
def test_factory_3_factory_kw_injections(self, providers):
"""Test factory with 3 keyword argument injections via factories."""
class A(object):
pass
class B(object):
pass
class C(object):
pass
class Test(object):
def __init__(self, a, b, c):
pass
a_factory = providers.Factory(A)
b_factory = providers.Factory(B)
c_factory = providers.Factory(C)
test_factory = providers.Factory(Test,
a=a_factory,
b=b_factory,
c=c_factory)
for x in xrange(int(5000000 * self.duration_factor)):
test_factory()
def test_factory_subcls_3_factory_subcls_kw_injections(self, providers):
"""Test factory with 3 keyword argument injections via factories."""
class MyFactory(providers.Factory):
pass
class A(object):
pass
class B(object):
pass
class C(object):
pass
class Test(object):
def __init__(self, a, b, c):
pass
a_factory = MyFactory(A)
b_factory = MyFactory(B)
c_factory = MyFactory(C)
test_factory = MyFactory(Test,
a=a_factory,
b=b_factory,
c=c_factory)
for x in xrange(int(5000000 * self.duration_factor)):
test_factory()
# def test_singleton(self, providers):
# """Test factory with 3 keyword argument injections via factories."""
# class Test(object):
# def __init__(self):
# pass
#
# test_factory = providers.Singleton(Test)
# for x in xrange(int(5000000 * self.duration_factor)):
# test_factory()
#
# def test_singleton_subcls(self, providers):
# """Test factory with 3 keyword argument injections via factories."""
# class MySingleton(providers.Singleton):
# pass
#
# class Test(object):
# pass
#
# test_factory = MySingleton(Test)
# for x in xrange(int(5000000 * self.duration_factor)):
# test_factory()
if __name__ == '__main__':
tester = Tester(
provider_modules=[
dependency_injector.providers,
],
duration_factor=1)
tester.run()

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

@ -80,23 +80,113 @@ class ProviderTests(unittest.TestCase):
self.assertEqual(self.provider.overridden, tuple())
def test_deepcopy(self):
provider = providers.Provider()
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.Provider)
def test_deepcopy_from_memo(self):
provider = providers.Provider()
provider_copy_memo = providers.Provider()
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_overridden(self):
provider = providers.Provider()
overriding_provider = providers.Provider()
provider.override(overriding_provider)
provider_copy = providers.deepcopy(provider)
overriding_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.Provider)
self.assertIsNot(overriding_provider, overriding_provider_copy)
self.assertIsInstance(overriding_provider_copy, providers.Provider)
def test_repr(self):
self.assertEqual(repr(self.provider),
'<dependency_injector.providers.base.'
'Provider() at {0}>'.format(hex(id(self.provider))))
class ObjectProviderTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(providers.is_provider(providers.Object(object())))
def test_call_object_provider(self):
obj = object()
self.assertIs(providers.Object(obj)(), obj)
def test_call_overridden_object_provider(self):
obj1 = object()
obj2 = object()
provider = providers.Object(obj1)
provider.override(providers.Object(obj2))
self.assertIs(provider(), obj2)
def test_deepcopy(self):
provider = providers.Object(1)
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.Object)
def test_deepcopy_from_memo(self):
provider = providers.Object(1)
provider_copy_memo = providers.Provider()
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_overridden(self):
provider = providers.Object(1)
overriding_provider = providers.Provider()
provider.override(overriding_provider)
provider_copy = providers.deepcopy(provider)
overriding_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.Object)
self.assertIsNot(overriding_provider, overriding_provider_copy)
self.assertIsInstance(overriding_provider_copy, providers.Provider)
def test_repr(self):
some_object = object()
provider = providers.Object(some_object)
self.assertEqual(repr(provider),
'<dependency_injector.providers.base.'
'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)
self.delegate = providers.Delegate(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())
self.assertRaises(errors.Error, providers.Delegate, object())
def test_call(self):
delegated1 = self.delegate()
@ -119,7 +209,7 @@ class ExternalDependencyTests(unittest.TestCase):
self.provider = providers.ExternalDependency(instance_of=list)
def test_init_with_not_class(self):
self.assertRaises(errors.Error, providers.ExternalDependency, object())
self.assertRaises(TypeError, providers.ExternalDependency, object())
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.provider))
@ -135,6 +225,38 @@ class ExternalDependencyTests(unittest.TestCase):
def test_call_not_overridden(self):
self.assertRaises(errors.Error, self.provider)
def test_deepcopy(self):
provider = providers.ExternalDependency(int)
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.ExternalDependency)
def test_deepcopy_from_memo(self):
provider = providers.ExternalDependency(int)
provider_copy_memo = providers.Provider()
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_overridden(self):
provider = providers.ExternalDependency(int)
overriding_provider = providers.Provider()
provider.override(overriding_provider)
provider_copy = providers.deepcopy(provider)
overriding_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIsInstance(provider, providers.ExternalDependency)
self.assertIsNot(overriding_provider, overriding_provider_copy)
self.assertIsInstance(overriding_provider_copy, providers.Provider)
def test_repr(self):
self.assertEqual(repr(self.provider),
'<dependency_injector.providers.base.'

91
tests/unit/providers/test_callable.py
View File

@ -1,91 +0,0 @@
"""Dependency injector callable providers unit tests."""
import unittest2 as unittest
from dependency_injector import (
providers,
errors,
)
class CallableTests(unittest.TestCase):
def example(self, arg1, arg2, arg3, arg4):
return arg1, arg2, arg3, arg4
def test_init_with_callable(self):
self.assertTrue(providers.Callable(self.example))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.Callable, 123)
def test_call(self):
provider = providers.Callable(lambda: True)
self.assertTrue(provider())
def test_call_with_positional_args(self):
provider = providers.Callable(self.example,
1, 2, 3, 4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_keyword_args(self):
provider = providers.Callable(self.example,
arg1=1, arg2=2, arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_positional_and_keyword_args(self):
provider = providers.Callable(self.example,
1, 2,
arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_context_args(self):
provider = providers.Callable(self.example, 1, 2)
self.assertTupleEqual(provider(3, 4), (1, 2, 3, 4))
def test_call_with_context_kwargs(self):
provider = providers.Callable(self.example, arg1=1)
self.assertTupleEqual(provider(arg2=2, arg3=3, arg4=4), (1, 2, 3, 4))
def test_call_with_context_args_and_kwargs(self):
provider = providers.Callable(self.example, 1)
self.assertTupleEqual(provider(2, arg3=3, arg4=4), (1, 2, 3, 4))
def test_fluent_interface(self):
provider = providers.Singleton(self.example) \
.add_args(1, 2) \
.add_kwargs(arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_overridden(self):
provider = providers.Callable(self.example)
provider.override(providers.Object((4, 3, 2, 1)))
provider.override(providers.Object((1, 2, 3, 4)))
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_repr(self):
provider = providers.Callable(self.example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.callable.'
'Callable({0}) at {1}>'.format(
repr(self.example),
hex(id(provider))))
class DelegatedCallableTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedCallable(len),
providers.Callable)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.DelegatedCallable(len)))
def test_is_delegated_provider(self):
provider = providers.DelegatedCallable(len)
self.assertIs(provider.provide_injection(), provider)

191
tests/unit/providers/test_callables.py Normal file
View File

@ -0,0 +1,191 @@
"""Dependency injector callable providers unit tests."""
import unittest2 as unittest
from dependency_injector import (
providers,
errors,
)
class CallableTests(unittest.TestCase):
def example(self, arg1, arg2, arg3, arg4):
return arg1, arg2, arg3, arg4
def test_init_with_callable(self):
self.assertTrue(providers.Callable(self.example))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.Callable, 123)
def test_call(self):
provider = providers.Callable(lambda: True)
self.assertTrue(provider())
def test_call_with_positional_args(self):
provider = providers.Callable(self.example,
1, 2, 3, 4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_keyword_args(self):
provider = providers.Callable(self.example,
arg1=1, arg2=2, arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_positional_and_keyword_args(self):
provider = providers.Callable(self.example,
1, 2,
arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_call_with_context_args(self):
provider = providers.Callable(self.example, 1, 2)
self.assertTupleEqual(provider(3, 4), (1, 2, 3, 4))
def test_call_with_context_kwargs(self):
provider = providers.Callable(self.example, arg1=1)
self.assertTupleEqual(provider(arg2=2, arg3=3, arg4=4), (1, 2, 3, 4))
def test_call_with_context_args_and_kwargs(self):
provider = providers.Callable(self.example, 1)
self.assertTupleEqual(provider(2, arg3=3, arg4=4), (1, 2, 3, 4))
def test_fluent_interface(self):
provider = providers.Singleton(self.example) \
.add_args(1, 2) \
.add_kwargs(arg3=3, arg4=4)
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_set_args(self):
provider = providers.Callable(self.example) \
.add_args(1, 2) \
.set_args(3, 4)
self.assertEqual(provider.args, tuple([3, 4]))
def test_set_kwargs(self):
provider = providers.Callable(self.example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.set_kwargs(init_arg3=4, init_arg4=5)
self.assertEqual(provider.kwargs, dict(init_arg3=4, init_arg4=5))
def test_clear_args(self):
provider = providers.Callable(self.example) \
.add_args(1, 2) \
.clear_args()
self.assertEqual(provider.args, tuple())
def test_clear_kwargs(self):
provider = providers.Callable(self.example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.clear_kwargs()
self.assertEqual(provider.kwargs, dict())
def test_call_overridden(self):
provider = providers.Callable(self.example)
provider.override(providers.Object((4, 3, 2, 1)))
provider.override(providers.Object((1, 2, 3, 4)))
self.assertTupleEqual(provider(), (1, 2, 3, 4))
def test_deepcopy(self):
provider = providers.Callable(self.example)
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.provides, provider_copy.provides)
self.assertIsInstance(provider, providers.Callable)
def test_deepcopy_from_memo(self):
provider = providers.Callable(self.example)
provider_copy_memo = providers.Callable(self.example)
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_args(self):
provider = providers.Callable(self.example)
dependent_provider1 = providers.Callable(list)
dependent_provider2 = providers.Callable(dict)
provider.add_args(dependent_provider1, dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.args[0]
dependent_provider_copy2 = provider_copy.args[1]
self.assertNotEqual(provider.args, provider_copy.args)
self.assertIs(dependent_provider1.provides,
dependent_provider_copy1.provides)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.provides,
dependent_provider_copy2.provides)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_kwargs(self):
provider = providers.Callable(self.example)
dependent_provider1 = providers.Callable(list)
dependent_provider2 = providers.Callable(dict)
provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.kwargs['a1']
dependent_provider_copy2 = provider_copy.kwargs['a2']
self.assertNotEqual(provider.kwargs, provider_copy.kwargs)
self.assertIs(dependent_provider1.provides,
dependent_provider_copy1.provides)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.provides,
dependent_provider_copy2.provides)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_overridden(self):
provider = providers.Callable(self.example)
object_provider = providers.Object(object())
provider.override(object_provider)
provider_copy = providers.deepcopy(provider)
object_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.provides, provider_copy.provides)
self.assertIsInstance(provider, providers.Callable)
self.assertIsNot(object_provider, object_provider_copy)
self.assertIsInstance(object_provider_copy, providers.Object)
def test_repr(self):
provider = providers.Callable(self.example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.callables.'
'Callable({0}) at {1}>'.format(
repr(self.example),
hex(id(provider))))
class DelegatedCallableTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedCallable(len),
providers.Callable)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.DelegatedCallable(len)))
def test_is_delegated_provider(self):
provider = providers.DelegatedCallable(len)
self.assertTrue(providers.is_delegated(provider))

662
tests/unit/providers/test_creational.py
View File

@ -1,662 +0,0 @@
"""Dependency injector creational providers unit tests."""
import unittest2 as unittest
from dependency_injector import (
providers,
errors,
)
class Example(object):
def __init__(self, init_arg1=None, init_arg2=None, init_arg3=None,
init_arg4=None):
self.init_arg1 = init_arg1
self.init_arg2 = init_arg2
self.init_arg3 = init_arg3
self.init_arg4 = init_arg4
self.attribute1 = None
self.attribute2 = None
class FactoryTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(providers.is_provider(providers.Factory(Example)))
def test_init_with_callable(self):
self.assertTrue(providers.Factory(credits))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.Factory, 123)
def test_init_with_valid_provided_type(self):
class ExampleProvider(providers.Factory):
provided_type = Example
example_provider = ExampleProvider(Example, 1, 2)
self.assertIsInstance(example_provider(), Example)
def test_init_with_valid_provided_subtype(self):
class ExampleProvider(providers.Factory):
provided_type = Example
class NewExampe(Example):
pass
example_provider = ExampleProvider(NewExampe, 1, 2)
self.assertIsInstance(example_provider(), NewExampe)
def test_init_with_invalid_provided_type(self):
class ExampleProvider(providers.Factory):
provided_type = Example
with self.assertRaises(errors.Error):
ExampleProvider(list)
def test_call(self):
provider = providers.Factory(Example)
instance1 = provider()
instance2 = provider()
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_args(self):
provider = providers.Factory(Example, 'i1', 'i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_keyword_args(self):
provider = providers.Factory(Example, init_arg1='i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_and_keyword_args(self):
provider = providers.Factory(Example, 'i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_attributes(self):
provider = providers.Factory(Example)
provider.add_attributes(attribute1='a1', attribute2='a2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.attribute1, 'a1')
self.assertEqual(instance1.attribute2, 'a2')
self.assertEqual(instance2.attribute1, 'a1')
self.assertEqual(instance2.attribute2, 'a2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_context_args(self):
provider = providers.Factory(Example, 11, 22)
instance = provider(33, 44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_call_with_context_kwargs(self):
provider = providers.Factory(Example, init_arg1=1)
instance1 = provider(init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
instance2 = provider(init_arg1=11, init_arg2=22)
self.assertEqual(instance2.init_arg1, 11)
self.assertEqual(instance2.init_arg2, 22)
def test_call_with_context_args_and_kwargs(self):
provider = providers.Factory(Example, 11)
instance = provider(22, init_arg3=33, init_arg4=44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_fluent_interface(self):
provider = providers.Factory(Example) \
.add_args(1, 2) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.add_attributes(attribute1=5, attribute2=6)
instance = provider()
self.assertEqual(instance.init_arg1, 1)
self.assertEqual(instance.init_arg2, 2)
self.assertEqual(instance.init_arg3, 3)
self.assertEqual(instance.init_arg4, 4)
self.assertEqual(instance.attribute1, 5)
self.assertEqual(instance.attribute2, 6)
def test_call_overridden(self):
provider = providers.Factory(Example)
overriding_provider1 = providers.Factory(dict)
overriding_provider2 = providers.Factory(list)
provider.override(overriding_provider1)
provider.override(overriding_provider2)
instance1 = provider()
instance2 = provider()
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, list)
self.assertIsInstance(instance2, list)
def test_repr(self):
provider = providers.Factory(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.creational.'
'Factory({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedFactoryTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedFactory(object),
providers.Factory)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.DelegatedFactory(object)))
def test_is_delegated_provider(self):
provider = providers.DelegatedFactory(object)
self.assertIs(provider.provide_injection(), provider)
class SingletonTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(providers.is_provider(providers.Singleton(Example)))
def test_init_with_callable(self):
self.assertTrue(providers.Singleton(credits))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.Singleton, 123)
def test_init_with_valid_provided_type(self):
class ExampleProvider(providers.Singleton):
provided_type = Example
example_provider = ExampleProvider(Example, 1, 2)
self.assertIsInstance(example_provider(), Example)
def test_init_with_valid_provided_subtype(self):
class ExampleProvider(providers.Singleton):
provided_type = Example
class NewExampe(Example):
pass
example_provider = ExampleProvider(NewExampe, 1, 2)
self.assertIsInstance(example_provider(), NewExampe)
def test_init_with_invalid_provided_type(self):
class ExampleProvider(providers.Singleton):
provided_type = Example
with self.assertRaises(errors.Error):
ExampleProvider(list)
def test_call(self):
provider = providers.Singleton(Example)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_args(self):
provider = providers.Singleton(Example, 'i1', 'i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_keyword_args(self):
provider = providers.Singleton(Example, init_arg1='i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_and_keyword_args(self):
provider = providers.Singleton(Example, 'i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_attributes(self):
provider = providers.Singleton(Example)
provider.add_attributes(attribute1='a1', attribute2='a2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.attribute1, 'a1')
self.assertEqual(instance1.attribute2, 'a2')
self.assertEqual(instance2.attribute1, 'a1')
self.assertEqual(instance2.attribute2, 'a2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_context_args(self):
provider = providers.Singleton(Example)
instance = provider(11, 22)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
def test_call_with_context_kwargs(self):
provider = providers.Singleton(Example, init_arg1=1)
instance1 = provider(init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
# Instance is created earlier
instance1 = provider(init_arg1=11, init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
def test_call_with_context_args_and_kwargs(self):
provider = providers.Singleton(Example, 11)
instance = provider(22, init_arg3=33, init_arg4=44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_fluent_interface(self):
provider = providers.Singleton(Example) \
.add_args(1, 2) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.add_attributes(attribute1=5, attribute2=6)
instance = provider()
self.assertEqual(instance.init_arg1, 1)
self.assertEqual(instance.init_arg2, 2)
self.assertEqual(instance.init_arg3, 3)
self.assertEqual(instance.init_arg4, 4)
self.assertEqual(instance.attribute1, 5)
self.assertEqual(instance.attribute2, 6)
def test_call_overridden(self):
provider = providers.Singleton(Example)
overriding_provider1 = providers.Singleton(dict)
overriding_provider2 = providers.Singleton(object)
provider.override(overriding_provider1)
provider.override(overriding_provider2)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, object)
self.assertIsInstance(instance2, object)
def test_reset(self):
provider = providers.Singleton(object)
instance1 = provider()
self.assertIsInstance(instance1, object)
provider.reset()
instance2 = provider()
self.assertIsInstance(instance1, object)
self.assertIsNot(instance1, instance2)
def test_repr(self):
provider = providers.Singleton(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.creational.'
'Singleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedSingletonTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedSingleton(object),
providers.Singleton)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.DelegatedSingleton(object)))
def test_is_delegated_provider(self):
provider = providers.DelegatedSingleton(object)
self.assertIs(provider.provide_injection(), provider)
class ThreadLocalSingletonTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.ThreadLocalSingleton(Example)))
def test_init_with_callable(self):
self.assertTrue(providers.ThreadLocalSingleton(credits))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.ThreadLocalSingleton, 123)
def test_init_with_valid_provided_type(self):
class ExampleProvider(providers.ThreadLocalSingleton):
provided_type = Example
example_provider = ExampleProvider(Example, 1, 2)
self.assertIsInstance(example_provider(), Example)
def test_init_with_valid_provided_subtype(self):
class ExampleProvider(providers.ThreadLocalSingleton):
provided_type = Example
class NewExampe(Example):
pass
example_provider = ExampleProvider(NewExampe, 1, 2)
self.assertIsInstance(example_provider(), NewExampe)
def test_init_with_invalid_provided_type(self):
class ExampleProvider(providers.ThreadLocalSingleton):
provided_type = Example
with self.assertRaises(errors.Error):
ExampleProvider(list)
def test_call(self):
provider = providers.ThreadLocalSingleton(Example)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_args(self):
provider = providers.ThreadLocalSingleton(Example, 'i1', 'i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_keyword_args(self):
provider = providers.ThreadLocalSingleton(Example,
init_arg1='i1',
init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_and_keyword_args(self):
provider = providers.ThreadLocalSingleton(Example,
'i1',
init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_attributes(self):
provider = providers.ThreadLocalSingleton(Example)
provider.add_attributes(attribute1='a1', attribute2='a2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.attribute1, 'a1')
self.assertEqual(instance1.attribute2, 'a2')
self.assertEqual(instance2.attribute1, 'a1')
self.assertEqual(instance2.attribute2, 'a2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_context_args(self):
provider = providers.ThreadLocalSingleton(Example)
instance = provider(11, 22)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
def test_call_with_context_kwargs(self):
provider = providers.ThreadLocalSingleton(Example, init_arg1=1)
instance1 = provider(init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
# Instance is created earlier
instance1 = provider(init_arg1=11, init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
def test_call_with_context_args_and_kwargs(self):
provider = providers.ThreadLocalSingleton(Example, 11)
instance = provider(22, init_arg3=33, init_arg4=44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_fluent_interface(self):
provider = providers.ThreadLocalSingleton(Example) \
.add_args(1, 2) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.add_attributes(attribute1=5, attribute2=6)
instance = provider()
self.assertEqual(instance.init_arg1, 1)
self.assertEqual(instance.init_arg2, 2)
self.assertEqual(instance.init_arg3, 3)
self.assertEqual(instance.init_arg4, 4)
self.assertEqual(instance.attribute1, 5)
self.assertEqual(instance.attribute2, 6)
def test_call_overridden(self):
provider = providers.ThreadLocalSingleton(Example)
overriding_provider1 = providers.ThreadLocalSingleton(dict)
overriding_provider2 = providers.ThreadLocalSingleton(object)
provider.override(overriding_provider1)
provider.override(overriding_provider2)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, object)
self.assertIsInstance(instance2, object)
def test_reset(self):
provider = providers.ThreadLocalSingleton(object)
instance1 = provider()
self.assertIsInstance(instance1, object)
provider.reset()
instance2 = provider()
self.assertIsInstance(instance1, object)
self.assertIsNot(instance1, instance2)
def test_repr(self):
provider = providers.ThreadLocalSingleton(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.creational.'
'ThreadLocalSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedThreadLocalSingletonTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedThreadLocalSingleton(object),
providers.ThreadLocalSingleton)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(
providers.DelegatedThreadLocalSingleton(object)))
def test_is_delegated_provider(self):
provider = providers.DelegatedThreadLocalSingleton(object)
self.assertIs(provider.provide_injection(), provider)
class FactoryAsDecoratorTests(unittest.TestCase):
def test_decoration_and_overriding(self):
@providers.Factory
class AuthService(object):
pass
@providers.override(AuthService)
@providers.Factory
class ExtAuthService(AuthService.cls):
pass
auth_service = AuthService()
self.assertIsInstance(auth_service, ExtAuthService.cls)

346
tests/unit/providers/test_factories.py Normal file
View File

@ -0,0 +1,346 @@
"""Dependency injector factory providers unit tests."""
import unittest2 as unittest
from dependency_injector import (
providers,
errors,
)
class Example(object):
def __init__(self, init_arg1=None, init_arg2=None, init_arg3=None,
init_arg4=None):
self.init_arg1 = init_arg1
self.init_arg2 = init_arg2
self.init_arg3 = init_arg3
self.init_arg4 = init_arg4
self.attribute1 = None
self.attribute2 = None
class FactoryTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(providers.is_provider(providers.Factory(Example)))
def test_init_with_callable(self):
self.assertTrue(providers.Factory(credits))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, providers.Factory, 123)
def test_init_with_valid_provided_type(self):
class ExampleProvider(providers.Factory):
provided_type = Example
example_provider = ExampleProvider(Example, 1, 2)
self.assertIsInstance(example_provider(), Example)
def test_init_with_valid_provided_subtype(self):
class ExampleProvider(providers.Factory):
provided_type = Example
class NewExampe(Example):
pass
example_provider = ExampleProvider(NewExampe, 1, 2)
self.assertIsInstance(example_provider(), NewExampe)
def test_init_with_invalid_provided_type(self):
class ExampleProvider(providers.Factory):
provided_type = Example
with self.assertRaises(errors.Error):
ExampleProvider(list)
def test_call(self):
provider = providers.Factory(Example)
instance1 = provider()
instance2 = provider()
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_args(self):
provider = providers.Factory(Example, 'i1', 'i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_keyword_args(self):
provider = providers.Factory(Example, init_arg1='i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_and_keyword_args(self):
provider = providers.Factory(Example, 'i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_attributes(self):
provider = providers.Factory(Example)
provider.add_attributes(attribute1='a1', attribute2='a2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.attribute1, 'a1')
self.assertEqual(instance1.attribute2, 'a2')
self.assertEqual(instance2.attribute1, 'a1')
self.assertEqual(instance2.attribute2, 'a2')
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_context_args(self):
provider = providers.Factory(Example, 11, 22)
instance = provider(33, 44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_call_with_context_kwargs(self):
provider = providers.Factory(Example, init_arg1=1)
instance1 = provider(init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
instance2 = provider(init_arg1=11, init_arg2=22)
self.assertEqual(instance2.init_arg1, 11)
self.assertEqual(instance2.init_arg2, 22)
def test_call_with_context_args_and_kwargs(self):
provider = providers.Factory(Example, 11)
instance = provider(22, init_arg3=33, init_arg4=44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_fluent_interface(self):
provider = providers.Factory(Example) \
.add_args(1, 2) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.add_attributes(attribute1=5, attribute2=6)
instance = provider()
self.assertEqual(instance.init_arg1, 1)
self.assertEqual(instance.init_arg2, 2)
self.assertEqual(instance.init_arg3, 3)
self.assertEqual(instance.init_arg4, 4)
self.assertEqual(instance.attribute1, 5)
self.assertEqual(instance.attribute2, 6)
def test_set_args(self):
provider = providers.Factory(Example) \
.add_args(1, 2) \
.set_args(3, 4)
self.assertEqual(provider.args, tuple([3, 4]))
def test_set_kwargs(self):
provider = providers.Factory(Example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.set_kwargs(init_arg3=4, init_arg4=5)
self.assertEqual(provider.kwargs, dict(init_arg3=4, init_arg4=5))
def test_set_attributes(self):
provider = providers.Factory(Example) \
.add_attributes(attribute1=5, attribute2=6) \
.set_attributes(attribute1=6, attribute2=7)
self.assertEqual(provider.attributes, dict(attribute1=6, attribute2=7))
def test_clear_args(self):
provider = providers.Factory(Example) \
.add_args(1, 2) \
.clear_args()
self.assertEqual(provider.args, tuple())
def test_clear_kwargs(self):
provider = providers.Factory(Example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.clear_kwargs()
self.assertEqual(provider.kwargs, dict())
def test_clear_attributes(self):
provider = providers.Factory(Example) \
.add_attributes(attribute1=5, attribute2=6) \
.clear_attributes()
self.assertEqual(provider.attributes, dict())
def test_call_overridden(self):
provider = providers.Factory(Example)
overriding_provider1 = providers.Factory(dict)
overriding_provider2 = providers.Factory(list)
provider.override(overriding_provider1)
provider.override(overriding_provider2)
instance1 = provider()
instance2 = provider()
self.assertIsNot(instance1, instance2)
self.assertIsInstance(instance1, list)
self.assertIsInstance(instance2, list)
def test_deepcopy(self):
provider = providers.Factory(Example)
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.cls, provider_copy.cls)
self.assertIsInstance(provider, providers.Factory)
def test_deepcopy_from_memo(self):
provider = providers.Factory(Example)
provider_copy_memo = providers.Factory(Example)
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_args(self):
provider = providers.Factory(Example)
dependent_provider1 = providers.Factory(list)
dependent_provider2 = providers.Factory(dict)
provider.add_args(dependent_provider1, dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.args[0]
dependent_provider_copy2 = provider_copy.args[1]
self.assertNotEqual(provider.args, provider_copy.args)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_kwargs(self):
provider = providers.Factory(Example)
dependent_provider1 = providers.Factory(list)
dependent_provider2 = providers.Factory(dict)
provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.kwargs['a1']
dependent_provider_copy2 = provider_copy.kwargs['a2']
self.assertNotEqual(provider.kwargs, provider_copy.kwargs)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_attributes(self):
provider = providers.Factory(Example)
dependent_provider1 = providers.Factory(list)
dependent_provider2 = providers.Factory(dict)
provider.add_attributes(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.attributes['a1']
dependent_provider_copy2 = provider_copy.attributes['a2']
self.assertNotEqual(provider.attributes, provider_copy.attributes)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_overridden(self):
provider = providers.Factory(Example)
object_provider = providers.Object(object())
provider.override(object_provider)
provider_copy = providers.deepcopy(provider)
object_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.cls, provider_copy.cls)
self.assertIsInstance(provider, providers.Factory)
self.assertIsNot(object_provider, object_provider_copy)
self.assertIsInstance(object_provider_copy, providers.Object)
def test_repr(self):
provider = providers.Factory(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.factories.'
'Factory({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedFactoryTests(unittest.TestCase):
def test_inheritance(self):
self.assertIsInstance(providers.DelegatedFactory(object),
providers.Factory)
def test_is_provider(self):
self.assertTrue(
providers.is_provider(providers.DelegatedFactory(object)))
def test_is_delegated_provider(self):
self.assertTrue(
providers.is_delegated(providers.DelegatedFactory(object)))

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

@ -25,6 +25,33 @@ 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)
def test_deepcopy(self):
provider = providers.Factory(object)
injection = providers.PositionalInjection(provider)
injection_copy = providers.deepcopy(injection)
self.assertIsNot(injection_copy, injection)
self.assertIsNot(injection_copy.get_original_value(),
injection.get_original_value())
def test_deepcopy_memo(self):
provider = providers.Factory(object)
injection = providers.PositionalInjection(provider)
injection_copy_orig = providers.PositionalInjection(provider)
injection_copy = providers.deepcopy(
injection, {id(injection): injection_copy_orig})
self.assertIs(injection_copy, injection_copy_orig)
self.assertIs(injection_copy.get_original_value(),
injection.get_original_value())
class NamedInjectionTests(unittest.TestCase):
@ -50,3 +77,30 @@ 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)
def test_deepcopy(self):
provider = providers.Factory(object)
injection = providers.NamedInjection('name', provider)
injection_copy = providers.deepcopy(injection)
self.assertIsNot(injection_copy, injection)
self.assertIsNot(injection_copy.get_original_value(),
injection.get_original_value())
def test_deepcopy_memo(self):
provider = providers.Factory(object)
injection = providers.NamedInjection('name', provider)
injection_copy_orig = providers.NamedInjection('name', provider)
injection_copy = providers.deepcopy(
injection, {id(injection): injection_copy_orig})
self.assertIs(injection_copy, injection_copy_orig)
self.assertIs(injection_copy.get_original_value(),
injection.get_original_value())

434
tests/unit/providers/test_singletons.py Normal file
View File

@ -0,0 +1,434 @@
"""Dependency injector singleton providers unit tests."""
import unittest2 as unittest
from dependency_injector import (
providers,
errors,
)
class Example(object):
def __init__(self, init_arg1=None, init_arg2=None, init_arg3=None,
init_arg4=None):
self.init_arg1 = init_arg1
self.init_arg2 = init_arg2
self.init_arg3 = init_arg3
self.init_arg4 = init_arg4
self.attribute1 = None
self.attribute2 = None
class _BaseSingletonTestCase(object):
singleton_cls = None
def test_is_provider(self):
self.assertTrue(providers.is_provider(self.singleton_cls(Example)))
def test_init_with_callable(self):
self.assertTrue(self.singleton_cls(credits))
def test_init_with_not_callable(self):
self.assertRaises(errors.Error, self.singleton_cls, 123)
def test_init_with_valid_provided_type(self):
class ExampleProvider(self.singleton_cls):
provided_type = Example
example_provider = ExampleProvider(Example, 1, 2)
self.assertIsInstance(example_provider(), Example)
def test_init_with_valid_provided_subtype(self):
class ExampleProvider(self.singleton_cls):
provided_type = Example
class NewExampe(Example):
pass
example_provider = ExampleProvider(NewExampe, 1, 2)
self.assertIsInstance(example_provider(), NewExampe)
def test_init_with_invalid_provided_type(self):
class ExampleProvider(self.singleton_cls):
provided_type = Example
with self.assertRaises(errors.Error):
ExampleProvider(list)
def test_call(self):
provider = self.singleton_cls(Example)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_args(self):
provider = self.singleton_cls(Example, 'i1', 'i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_keyword_args(self):
provider = self.singleton_cls(Example, init_arg1='i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_init_positional_and_keyword_args(self):
provider = self.singleton_cls(Example, 'i1', init_arg2='i2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.init_arg1, 'i1')
self.assertEqual(instance1.init_arg2, 'i2')
self.assertEqual(instance2.init_arg1, 'i1')
self.assertEqual(instance2.init_arg2, 'i2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_attributes(self):
provider = self.singleton_cls(Example)
provider.add_attributes(attribute1='a1', attribute2='a2')
instance1 = provider()
instance2 = provider()
self.assertEqual(instance1.attribute1, 'a1')
self.assertEqual(instance1.attribute2, 'a2')
self.assertEqual(instance2.attribute1, 'a1')
self.assertEqual(instance2.attribute2, 'a2')
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, Example)
self.assertIsInstance(instance2, Example)
def test_call_with_context_args(self):
provider = self.singleton_cls(Example)
instance = provider(11, 22)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
def test_call_with_context_kwargs(self):
provider = self.singleton_cls(Example, init_arg1=1)
instance1 = provider(init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
# Instance is created earlier
instance1 = provider(init_arg1=11, init_arg2=22)
self.assertEqual(instance1.init_arg1, 1)
self.assertEqual(instance1.init_arg2, 22)
def test_call_with_context_args_and_kwargs(self):
provider = self.singleton_cls(Example, 11)
instance = provider(22, init_arg3=33, init_arg4=44)
self.assertEqual(instance.init_arg1, 11)
self.assertEqual(instance.init_arg2, 22)
self.assertEqual(instance.init_arg3, 33)
self.assertEqual(instance.init_arg4, 44)
def test_fluent_interface(self):
provider = self.singleton_cls(Example) \
.add_args(1, 2) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.add_attributes(attribute1=5, attribute2=6)
instance = provider()
self.assertEqual(instance.init_arg1, 1)
self.assertEqual(instance.init_arg2, 2)
self.assertEqual(instance.init_arg3, 3)
self.assertEqual(instance.init_arg4, 4)
self.assertEqual(instance.attribute1, 5)
self.assertEqual(instance.attribute2, 6)
def test_set_args(self):
provider = self.singleton_cls(Example) \
.add_args(1, 2) \
.set_args(3, 4)
self.assertEqual(provider.args, tuple([3, 4]))
def test_set_kwargs(self):
provider = self.singleton_cls(Example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.set_kwargs(init_arg3=4, init_arg4=5)
self.assertEqual(provider.kwargs, dict(init_arg3=4, init_arg4=5))
def test_set_attributes(self):
provider = self.singleton_cls(Example) \
.add_attributes(attribute1=5, attribute2=6) \
.set_attributes(attribute1=6, attribute2=7)
self.assertEqual(provider.attributes, dict(attribute1=6, attribute2=7))
def test_clear_args(self):
provider = self.singleton_cls(Example) \
.add_args(1, 2) \
.clear_args()
self.assertEqual(provider.args, tuple())
def test_clear_kwargs(self):
provider = self.singleton_cls(Example) \
.add_kwargs(init_arg3=3, init_arg4=4) \
.clear_kwargs()
self.assertEqual(provider.kwargs, dict())
def test_clear_attributes(self):
provider = self.singleton_cls(Example) \
.add_attributes(attribute1=5, attribute2=6) \
.clear_attributes()
self.assertEqual(provider.attributes, dict())
def test_call_overridden(self):
provider = self.singleton_cls(Example)
overriding_provider1 = self.singleton_cls(dict)
overriding_provider2 = self.singleton_cls(list)
provider.override(overriding_provider1)
provider.override(overriding_provider2)
instance1 = provider()
instance2 = provider()
self.assertIs(instance1, instance2)
self.assertIsInstance(instance1, list)
self.assertIsInstance(instance2, list)
def test_deepcopy(self):
provider = self.singleton_cls(Example)
provider_copy = providers.deepcopy(provider)
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.cls, provider_copy.cls)
self.assertIsInstance(provider, self.singleton_cls)
def test_deepcopy_from_memo(self):
provider = self.singleton_cls(Example)
provider_copy_memo = self.singleton_cls(Example)
provider_copy = providers.deepcopy(
provider, memo={id(provider): provider_copy_memo})
self.assertIs(provider_copy, provider_copy_memo)
def test_deepcopy_args(self):
provider = self.singleton_cls(Example)
dependent_provider1 = self.singleton_cls(list)
dependent_provider2 = self.singleton_cls(dict)
provider.add_args(dependent_provider1, dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.args[0]
dependent_provider_copy2 = provider_copy.args[1]
self.assertNotEqual(provider.args, provider_copy.args)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_kwargs(self):
provider = self.singleton_cls(Example)
dependent_provider1 = self.singleton_cls(list)
dependent_provider2 = self.singleton_cls(dict)
provider.add_kwargs(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.kwargs['a1']
dependent_provider_copy2 = provider_copy.kwargs['a2']
self.assertNotEqual(provider.kwargs, provider_copy.kwargs)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_attributes(self):
provider = self.singleton_cls(Example)
dependent_provider1 = self.singleton_cls(list)
dependent_provider2 = self.singleton_cls(dict)
provider.add_attributes(a1=dependent_provider1, a2=dependent_provider2)
provider_copy = providers.deepcopy(provider)
dependent_provider_copy1 = provider_copy.attributes['a1']
dependent_provider_copy2 = provider_copy.attributes['a2']
self.assertNotEqual(provider.attributes, provider_copy.attributes)
self.assertIs(dependent_provider1.cls, dependent_provider_copy1.cls)
self.assertIsNot(dependent_provider1, dependent_provider_copy1)
self.assertIs(dependent_provider2.cls, dependent_provider_copy2.cls)
self.assertIsNot(dependent_provider2, dependent_provider_copy2)
def test_deepcopy_overridden(self):
provider = self.singleton_cls(Example)
object_provider = providers.Object(object())
provider.override(object_provider)
provider_copy = providers.deepcopy(provider)
object_provider_copy = provider_copy.overridden[0]
self.assertIsNot(provider, provider_copy)
self.assertIs(provider.cls, provider_copy.cls)
self.assertIsInstance(provider, self.singleton_cls)
self.assertIsNot(object_provider, object_provider_copy)
self.assertIsInstance(object_provider_copy, providers.Object)
def test_reset(self):
provider = self.singleton_cls(object)
instance1 = provider()
self.assertIsInstance(instance1, object)
provider.reset()
instance2 = provider()
self.assertIsInstance(instance1, object)
self.assertIsNot(instance1, instance2)
class SingletonTests(_BaseSingletonTestCase, unittest.TestCase):
singleton_cls = providers.Singleton
def test_repr(self):
provider = self.singleton_cls(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'Singleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedSingletonTests(_BaseSingletonTestCase, unittest.TestCase):
singleton_cls = providers.DelegatedSingleton
def test_is_delegated_provider(self):
provider = self.singleton_cls(object)
self.assertTrue(providers.is_delegated(provider))
def test_repr(self):
provider = self.singleton_cls(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'DelegatedSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class ThreadLocalSingletonTests(_BaseSingletonTestCase, unittest.TestCase):
singleton_cls = providers.ThreadLocalSingleton
def test_repr(self):
provider = providers.ThreadLocalSingleton(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'ThreadLocalSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedThreadLocalSingletonTests(_BaseSingletonTestCase,
unittest.TestCase):
singleton_cls = providers.DelegatedThreadLocalSingleton
def test_is_delegated_provider(self):
provider = self.singleton_cls(object)
self.assertTrue(providers.is_delegated(provider))
def test_repr(self):
provider = self.singleton_cls(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'DelegatedThreadLocalSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class ThreadSafeSingletonTests(_BaseSingletonTestCase, unittest.TestCase):
singleton_cls = providers.ThreadSafeSingleton
def test_repr(self):
provider = self.singleton_cls(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'ThreadSafeSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))
class DelegatedThreadSafeSingletonTests(_BaseSingletonTestCase,
unittest.TestCase):
singleton_cls = providers.DelegatedThreadSafeSingleton
def test_is_delegated_provider(self):
provider = self.singleton_cls(object)
self.assertTrue(providers.is_delegated(provider))
def test_repr(self):
provider = self.singleton_cls(Example)
self.assertEqual(repr(provider),
'<dependency_injector.providers.singletons.'
'DelegatedThreadSafeSingleton({0}) at {1}>'.format(
repr(Example),
hex(id(provider))))

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

@ -1,31 +0,0 @@
"""Dependency injector static providers unit tests."""
import unittest2 as unittest
from dependency_injector import providers
class ObjectProviderTests(unittest.TestCase):
def test_is_provider(self):
self.assertTrue(providers.is_provider(providers.Object(object())))
def test_call_object_provider(self):
obj = object()
self.assertIs(providers.Object(obj)(), obj)
def test_call_overridden_object_provider(self):
obj1 = object()
obj2 = object()
provider = providers.Object(obj1)
provider.override(providers.Object(obj2))
self.assertIs(provider(), obj2)
def test_repr(self):
some_object = object()
provider = providers.Object(some_object)
self.assertEqual(repr(provider),
'<dependency_injector.providers.base.'
'Object({0}) at {1}>'.format(
repr(some_object),
hex(id(provider))))