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

Split `catalogs` module into smaller modules

Browse Source
This commit is contained in:
Roman Mogilatov 2016-04-03 23:47:34 +03:00
parent 040d44d55b
commit 136b562a81
19 changed files with 1778 additions and 1658 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

895
dependency_injector/catalogs.py
View File

@ -1,895 +0,0 @@
"""Catalogs module."""
import six
from dependency_injector.errors import (
Error,
UndefinedProviderError,
)
from dependency_injector.utils import (
is_provider,
is_catalog,
is_declarative_catalog,
ensure_is_provider,
ensure_is_catalog_bundle,
)
@six.python_2_unicode_compatible
class CatalogBundle(object):
"""Bundle of catalog providers.
:py:class:`CatalogBundle` is a frozen, limited collection of catalog
providers. While catalog could be used as a centralized place for
particular providers group, such bundles of catalog providers can be used
for creating several frozen, limited scopes that could be passed to
different subsystems.
:py:class:`CatalogBundle` has API's parity with catalogs
(:py:class:`DeclarativeCatalog` or :py:class:`DynamicCatalog`) in terms of
retrieving the providers, but it is "frozen" in terms of modification
provider's list.
:py:class:`CatalogBundle` is considered to be dependable on catalogs
(:py:class:`DeclarativeCatalog` or :py:class:`DynamicCatalog`) entity by
its design.
.. py:attribute:: catalog
Bundle's catalog.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`
.. py:attribute:: providers
Dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
catalog = None
__IS_CATALOG_BUNDLE__ = True
__slots__ = ('providers', '__dict__')
@classmethod
def sub_cls_factory(cls, catalog):
"""Create bundle subclass for catalog.
:return: Subclass of :py:class:`CatalogBundle`.
:rtype: :py:class:`CatalogBundle`
"""
return type('BundleSubclass', (cls,), dict(catalog=catalog))
def __init__(self, *providers):
"""Initializer.
:param providers: Tuple of catalog's bundle providers.
:type providers: tuple[
:py:class:`dependency_injector.providers.Provider`]
"""
self.providers = dict((self.catalog.get_provider_bind_name(provider),
provider)
for provider in providers)
self.__dict__.update(self.providers)
super(CatalogBundle, self).__init__()
def get_provider(self, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
try:
return self.providers[name]
except KeyError:
raise Error('Provider "{0}" is not a part of {1}'.format(name,
self))
def has_provider(self, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return name in self.providers
def __getattr__(self, item):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
if item.startswith('__') and item.endswith('__'):
return super(CatalogBundle, self).__getattr__(item)
raise UndefinedProviderError('Provider "{0}" is not a part '
'of {1}'.format(item, self))
def __repr__(self):
"""Return string representation of catalog's bundle.
:rtype: str
"""
return '<{0}.Bundle({1})>'.format(
self.catalog.name, ', '.join(six.iterkeys(self.providers)))
__str__ = __repr__
@six.python_2_unicode_compatible
class DynamicCatalog(object):
"""Dynamic catalog of providers.
:py:class:`DynamicCatalog` is a catalog of providers that could be created
in application's runtime. It should cover most of the cases when list of
providers that would be included in catalog is non-deterministic in terms
of apllication code (catalog's structure could be determined just after
application will be started and will do some initial work, like parsing
list of catalog's providers from the configuration).
.. code-block:: python
services = DynamicCatalog(auth=providers.Factory(AuthService),
users=providers.Factory(UsersService))
users_service = services.users()
.. py:attribute:: Bundle
Catalog's bundle class.
:type: :py:class:`CatalogBundle`
.. py:attribute:: name
Catalog's name.
By default, it is catalog's module + catalog's class name.
:type: str
.. py:attribute:: providers
Dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: overridden_by
Tuple of overriding catalogs.
:type: tuple[
:py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`]
.. py:attribute:: provider_type
If provider type is defined, :py:class:`DynamicCatalog` checks that
all of its providers are instances of
:py:attr:`DynamicCatalog.provider_type`.
:type: type | None
"""
provider_type = None
__IS_CATALOG__ = True
__slots__ = ('name', 'providers', 'provider_names', 'overridden_by',
'Bundle')
def __init__(self, **providers):
"""Initializer.
:param providers: Dictionary of catalog providers.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
self.Bundle = CatalogBundle.sub_cls_factory(self)
self.name = '.'.join((self.__class__.__module__,
self.__class__.__name__))
self.providers = dict()
self.provider_names = dict()
self.overridden_by = tuple()
self.bind_providers(providers)
super(DynamicCatalog, self).__init__()
def is_bundle_owner(self, bundle):
"""Check if catalog is bundle owner.
:param bundle: Catalog's bundle instance.
:type bundle: :py:class:`CatalogBundle`
:rtype: bool
"""
return ensure_is_catalog_bundle(bundle) and bundle.catalog is self
def get_provider_bind_name(self, provider):
"""Return provider's name in catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider's name.
:rtype: str
"""
if not self.is_provider_bound(provider):
raise Error('Can not find bind name for {0} in catalog {1}'.format(
provider, self))
return self.provider_names[provider]
def is_provider_bound(self, provider):
"""Check if provider is bound to the catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:rtype: bool
"""
return provider in self.provider_names
def filter(self, provider_type):
"""Return dictionary of providers, that are instance of provided type.
:param provider_type: Provider's type.
:type provider_type: :py:class:`dependency_injector.providers.Provider`
:rtype: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return dict((name, provider)
for name, provider in six.iteritems(self.providers)
if isinstance(provider, provider_type))
@property
def is_overridden(self):
"""Read-only property that is set to ``True`` if catalog is overridden.
:rtype: bool
"""
return bool(self.overridden_by)
@property
def last_overriding(self):
"""Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
"""
return self.overridden_by[-1] if self.overridden_by else None
def override(self, overriding):
"""Override current catalog providers by overriding catalog providers.
:param overriding: Overriding catalog.
:type overriding: :py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`
:raise: :py:exc:`dependency_injector.errors.Error` if trying to
override catalog by itself
:rtype: None
"""
if overriding is self:
raise Error('Catalog {0} could not be overridden '
'with itself'.format(self))
self.overridden_by += (overriding,)
for name, provider in six.iteritems(overriding.providers):
self.get_provider(name).override(provider)
def reset_last_overriding(self):
"""Reset last overriding catalog.
:rtype: None
"""
if not self.is_overridden:
raise Error('Catalog {0} is not overridden'.format(self))
self.overridden_by = self.overridden_by[:-1]
for provider in six.itervalues(self.providers):
provider.reset_last_overriding()
def reset_override(self):
"""Reset all overridings for all catalog providers.
:rtype: None
"""
self.overridden_by = tuple()
for provider in six.itervalues(self.providers):
provider.reset_override()
def get_provider(self, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
try:
return self.providers[name]
except KeyError:
raise UndefinedProviderError('{0} has no provider with such '
'name - {1}'.format(self, name))
def bind_provider(self, name, provider):
"""Bind provider to catalog with specified name.
:param name: Name of the provider.
:type name: str
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
provider = ensure_is_provider(provider)
if (self.__class__.provider_type and
not isinstance(provider, self.__class__.provider_type)):
raise Error('{0} can contain only {1} instances'.format(
self, self.__class__.provider_type))
if name in self.providers:
raise Error('Catalog {0} already has provider with '
'such name - {1}'.format(self, name))
if provider in self.provider_names:
raise Error('Catalog {0} already has such provider '
'instance - {1}'.format(self, provider))
self.providers[name] = provider
self.provider_names[provider] = name
def bind_providers(self, providers):
"""Bind providers dictionary to catalog.
:param providers: Dictionary of providers, where key is a name
and value is a provider.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
for name, provider in six.iteritems(providers):
self.bind_provider(name, provider)
def has_provider(self, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return name in self.providers
def unbind_provider(self, name):
"""Remove provider binding.
:param name: Provider's name.
:type name: str
:rtype: None
"""
provider = self.get_provider(name)
del self.providers[name]
del self.provider_names[provider]
def __getattr__(self, name):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
return self.get_provider(name)
def __setattr__(self, name, value):
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
if is_provider(value):
return self.bind_provider(name, value)
return super(DynamicCatalog, self).__setattr__(name, value)
def __delattr__(self, name):
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
self.unbind_provider(name)
def __repr__(self):
"""Return Python representation of catalog.
:rtype: str
"""
return '<{0}({1})>'.format(self.name,
', '.join(six.iterkeys(self.providers)))
__str__ = __repr__
@six.python_2_unicode_compatible
class DeclarativeCatalogMetaClass(type):
"""Declarative catalog meta class."""
def __new__(mcs, class_name, bases, attributes):
"""Declarative catalog class factory."""
cls_providers = tuple((name, provider)
for name, provider in six.iteritems(attributes)
if is_provider(provider))
inherited_providers = tuple((name, provider)
for base in bases if is_catalog(base)
for name, provider in six.iteritems(
base.providers))
providers = cls_providers + inherited_providers
cls = type.__new__(mcs, class_name, bases, attributes)
if cls.provider_type:
cls._catalog = type('DynamicCatalog',
(DynamicCatalog,),
dict(provider_type=cls.provider_type))()
else:
cls._catalog = DynamicCatalog()
cls._catalog.name = '.'.join((cls.__module__, cls.__name__))
cls._catalog.bind_providers(dict(providers))
cls.cls_providers = dict(cls_providers)
cls.inherited_providers = dict(inherited_providers)
cls.Bundle = cls._catalog.Bundle
return cls
@property
def name(cls):
"""Read-only property that represents catalog's name.
Catalog's name is catalog's module + catalog's class name.
:type: str
"""
return cls._catalog.name
@property
def providers(cls):
"""Read-only dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return cls._catalog.providers
@property
def overridden_by(cls):
"""Tuple of overriding catalogs.
:type: tuple[
:py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`]
"""
return cls._catalog.overridden_by
@property
def is_overridden(cls):
"""Read-only property that is set to ``True`` if catalog is overridden.
:rtype: bool
"""
return cls._catalog.is_overridden
@property
def last_overriding(cls):
"""Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
"""
return cls._catalog.last_overriding
def __getattr__(cls, name):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
raise UndefinedProviderError('There is no provider "{0}" in '
'catalog {1}'.format(name, cls))
def __setattr__(cls, name, value):
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
if is_provider(value):
setattr(cls._catalog, name, value)
return super(DeclarativeCatalogMetaClass, cls).__setattr__(name, value)
def __delattr__(cls, name):
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
if is_provider(getattr(cls, name)):
delattr(cls._catalog, name)
return super(DeclarativeCatalogMetaClass, cls).__delattr__(name)
def __repr__(cls):
"""Return string representation of the catalog.
:rtype: str
"""
return '<{0}({1})>'.format(cls.name,
', '.join(six.iterkeys(cls.providers)))
__str__ = __repr__
@six.add_metaclass(DeclarativeCatalogMetaClass)
class DeclarativeCatalog(object):
"""Declarative catalog of providers.
:py:class:`DeclarativeCatalog` is a catalog of providers that could be
defined in declarative manner. It should cover most of the cases when list
of providers that would be included in catalog is deterministic (catalog
will not change its structure in runtime).
.. code-block:: python
class Services(DeclarativeCatalog):
auth = providers.Factory(AuthService)
users = providers.Factory(UsersService)
users_service = Services.users()
.. py:attribute:: Bundle
Catalog's bundle class.
:type: :py:class:`CatalogBundle`
.. py:attribute:: name
Read-only property that represents catalog's name.
Catalog's name is catalog's module + catalog's class name.
:type: str
.. py:attribute:: cls_providers
Read-only dictionary of current catalog providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: inherited_providers
Read-only dictionary of inherited providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: providers
Read-only dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: overridden_by
Tuple of overriding catalogs.
:type: tuple[:py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`]
.. py:attribute:: is_overridden
Read-only property that is set to ``True`` if catalog is overridden.
:type: bool
.. py:attribute:: is_overridden
Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
.. py:attribute:: provider_type
If provider type is defined, :py:class:`DeclarativeCatalog` checks that
all of its providers are instances of
:py:attr:`DeclarativeCatalog.provider_type`.
:type: type | None
"""
Bundle = CatalogBundle
name = str()
cls_providers = dict()
inherited_providers = dict()
providers = dict()
overridden_by = tuple()
is_overridden = bool
last_overriding = None
provider_type = None
_catalog = DynamicCatalog
__IS_CATALOG__ = True
@classmethod
def is_bundle_owner(cls, bundle):
"""Check if catalog is bundle owner.
:param bundle: Catalog's bundle instance.
:type bundle: :py:class:`CatalogBundle`
:rtype: bool
"""
return cls._catalog.is_bundle_owner(bundle)
@classmethod
def get_provider_bind_name(cls, provider):
"""Return provider's name in catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider's name.
:rtype: str
"""
return cls._catalog.get_provider_bind_name(provider)
@classmethod
def is_provider_bound(cls, provider):
"""Check if provider is bound to the catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:rtype: bool
"""
return cls._catalog.is_provider_bound(provider)
@classmethod
def filter(cls, provider_type):
"""Return dictionary of providers, that are instance of provided type.
:param provider_type: Provider's type.
:type provider_type: :py:class:`dependency_injector.providers.Provider`
:rtype: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return cls._catalog.filter(provider_type)
@classmethod
def override(cls, overriding):
"""Override current catalog providers by overriding catalog providers.
:param overriding: Overriding catalog.
:type overriding: :py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`
:raise: :py:exc:`dependency_injector.errors.Error` if trying to
override catalog by itself or its subclasses
:rtype: None
"""
if is_declarative_catalog(overriding) and issubclass(cls, overriding):
raise Error('Catalog {0} could not be overridden '
'with itself or its subclasses'.format(cls))
return cls._catalog.override(overriding)
@classmethod
def reset_last_overriding(cls):
"""Reset last overriding catalog.
:rtype: None
"""
cls._catalog.reset_last_overriding()
@classmethod
def reset_override(cls):
"""Reset all overridings for all catalog providers.
:rtype: None
"""
cls._catalog.reset_override()
@classmethod
def get_provider(cls, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
return cls._catalog.get_provider(name)
get = get_provider # Backward compatibility for versions < 0.11.*
@classmethod
def bind_provider(cls, name, provider):
"""Bind provider to catalog with specified name.
:param name: Name of the provider.
:type name: str
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
setattr(cls, name, provider)
@classmethod
def bind_providers(cls, providers):
"""Bind providers dictionary to catalog.
:param providers: Dictionary of providers, where key is a name
and value is a provider.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
for name, provider in six.iteritems(providers):
setattr(cls, name, provider)
@classmethod
def has_provider(cls, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return hasattr(cls, name)
has = has_provider # Backward compatibility for versions < 0.11.*
@classmethod
def unbind_provider(cls, name):
"""Remove provider binding.
:param name: Provider's name.
:type name: str
:rtype: None
"""
delattr(cls, name)
@classmethod
def __getattr__(cls, name): # pragma: no cover
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
raise NotImplementedError('Implementated in metaclass')
@classmethod
def __setattr__(cls, name, value): # pragma: no cover
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
raise NotImplementedError('Implementated in metaclass')
@classmethod
def __delattr__(cls, name): # pragma: no cover
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
raise NotImplementedError('Implementated in metaclass')
# Backward compatibility for versions < 0.11.*
AbstractCatalog = DeclarativeCatalog
def override(catalog):
""":py:class:`DeclarativeCatalog` overriding decorator.
:param catalog: Catalog that should be overridden by decorated catalog.
:type catalog: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`
:return: Declarative catalog's overriding decorator.
:rtype: callable(:py:class:`DeclarativeCatalog`)
"""
def decorator(overriding_catalog):
"""Overriding decorator."""
catalog.override(overriding_catalog)
return overriding_catalog
return decorator

37
dependency_injector/catalogs/__init__.py Normal file
View File

@ -0,0 +1,37 @@
"""Dependency injector catalogs package."""
from dependency_injector.catalogs.bundle import CatalogBundle
from dependency_injector.catalogs.dynamic import DynamicCatalog
from dependency_injector.catalogs.declarative import (
DeclarativeCatalogMetaClass,
DeclarativeCatalog,
AbstractCatalog,
)
def override(catalog):
""":py:class:`DeclarativeCatalog` overriding decorator.
:param catalog: Catalog that should be overridden by decorated catalog.
:type catalog: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`
:return: Declarative catalog's overriding decorator.
:rtype: callable(:py:class:`DeclarativeCatalog`)
"""
def decorator(overriding_catalog):
"""Overriding decorator."""
catalog.override(overriding_catalog)
return overriding_catalog
return decorator
__all__ = (
'CatalogBundle',
'DynamicCatalog',
'DeclarativeCatalogMetaClass',
'DeclarativeCatalog',
'AbstractCatalog',
'override',
)

118
dependency_injector/catalogs/bundle.py Normal file
View File

@ -0,0 +1,118 @@
"""Dependency injector catalogs bundle module."""
import six
from dependency_injector.errors import (
Error,
UndefinedProviderError,
)
@six.python_2_unicode_compatible
class CatalogBundle(object):
"""Bundle of catalog providers.
:py:class:`CatalogBundle` is a frozen, limited collection of catalog
providers. While catalog could be used as a centralized place for
particular providers group, such bundles of catalog providers can be used
for creating several frozen, limited scopes that could be passed to
different subsystems.
:py:class:`CatalogBundle` has API's parity with catalogs
(:py:class:`DeclarativeCatalog` or :py:class:`DynamicCatalog`) in terms of
retrieving the providers, but it is "frozen" in terms of modification
provider's list.
:py:class:`CatalogBundle` is considered to be dependable on catalogs
(:py:class:`DeclarativeCatalog` or :py:class:`DynamicCatalog`) entity by
its design.
.. py:attribute:: catalog
Bundle's catalog.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`
.. py:attribute:: providers
Dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
catalog = None
__IS_CATALOG_BUNDLE__ = True
__slots__ = ('providers', '__dict__')
@classmethod
def sub_cls_factory(cls, catalog):
"""Create bundle subclass for catalog.
:return: Subclass of :py:class:`CatalogBundle`.
:rtype: :py:class:`CatalogBundle`
"""
return type('BundleSubclass', (cls,), dict(catalog=catalog))
def __init__(self, *providers):
"""Initializer.
:param providers: Tuple of catalog's bundle providers.
:type providers: tuple[
:py:class:`dependency_injector.providers.Provider`]
"""
self.providers = dict((self.catalog.get_provider_bind_name(provider),
provider)
for provider in providers)
self.__dict__.update(self.providers)
super(CatalogBundle, self).__init__()
def get_provider(self, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
try:
return self.providers[name]
except KeyError:
raise Error('Provider "{0}" is not a part of {1}'.format(name,
self))
def has_provider(self, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return name in self.providers
def __getattr__(self, item):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
if item.startswith('__') and item.endswith('__'):
return super(CatalogBundle, self).__getattr__(item)
raise UndefinedProviderError('Provider "{0}" is not a part '
'of {1}'.format(item, self))
def __repr__(self):
"""Return string representation of catalog's bundle.
:rtype: str
"""
return '<{0}.Bundle({1})>'.format(
self.catalog.name, ', '.join(six.iterkeys(self.providers)))
__str__ = __repr__

453
dependency_injector/catalogs/declarative.py Normal file
View File

@ -0,0 +1,453 @@
"""Dependency injector declarative catalog module."""
import six
from dependency_injector.catalogs.dynamic import DynamicCatalog
from dependency_injector.catalogs.bundle import CatalogBundle
from dependency_injector.utils import (
is_provider,
is_catalog,
is_declarative_catalog,
)
from dependency_injector.errors import (
Error,
UndefinedProviderError,
)
@six.python_2_unicode_compatible
class DeclarativeCatalogMetaClass(type):
"""Declarative catalog meta class."""
def __new__(mcs, class_name, bases, attributes):
"""Declarative catalog class factory."""
cls_providers = tuple((name, provider)
for name, provider in six.iteritems(attributes)
if is_provider(provider))
inherited_providers = tuple((name, provider)
for base in bases if is_catalog(base)
for name, provider in six.iteritems(
base.providers))
providers = cls_providers + inherited_providers
cls = type.__new__(mcs, class_name, bases, attributes)
if cls.provider_type:
cls._catalog = type('DynamicCatalog',
(DynamicCatalog,),
dict(provider_type=cls.provider_type))()
else:
cls._catalog = DynamicCatalog()
cls._catalog.name = '.'.join((cls.__module__, cls.__name__))
cls._catalog.bind_providers(dict(providers))
cls.cls_providers = dict(cls_providers)
cls.inherited_providers = dict(inherited_providers)
cls.Bundle = cls._catalog.Bundle
return cls
@property
def name(cls):
"""Read-only property that represents catalog's name.
Catalog's name is catalog's module + catalog's class name.
:type: str
"""
return cls._catalog.name
@property
def providers(cls):
"""Read-only dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return cls._catalog.providers
@property
def overridden_by(cls):
"""Tuple of overriding catalogs.
:type: tuple[
:py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`]
"""
return cls._catalog.overridden_by
@property
def is_overridden(cls):
"""Read-only property that is set to ``True`` if catalog is overridden.
:rtype: bool
"""
return cls._catalog.is_overridden
@property
def last_overriding(cls):
"""Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
"""
return cls._catalog.last_overriding
def __getattr__(cls, name):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
raise UndefinedProviderError('There is no provider "{0}" in '
'catalog {1}'.format(name, cls))
def __setattr__(cls, name, value):
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
if is_provider(value):
setattr(cls._catalog, name, value)
return super(DeclarativeCatalogMetaClass, cls).__setattr__(name, value)
def __delattr__(cls, name):
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
if is_provider(getattr(cls, name)):
delattr(cls._catalog, name)
return super(DeclarativeCatalogMetaClass, cls).__delattr__(name)
def __repr__(cls):
"""Return string representation of the catalog.
:rtype: str
"""
return '<{0}({1})>'.format(cls.name,
', '.join(six.iterkeys(cls.providers)))
__str__ = __repr__
@six.add_metaclass(DeclarativeCatalogMetaClass)
class DeclarativeCatalog(object):
"""Declarative catalog of providers.
:py:class:`DeclarativeCatalog` is a catalog of providers that could be
defined in declarative manner. It should cover most of the cases when list
of providers that would be included in catalog is deterministic (catalog
will not change its structure in runtime).
.. code-block:: python
class Services(DeclarativeCatalog):
auth = providers.Factory(AuthService)
users = providers.Factory(UsersService)
users_service = Services.users()
.. py:attribute:: Bundle
Catalog's bundle class.
:type: :py:class:`CatalogBundle`
.. py:attribute:: name
Read-only property that represents catalog's name.
Catalog's name is catalog's module + catalog's class name.
:type: str
.. py:attribute:: cls_providers
Read-only dictionary of current catalog providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: inherited_providers
Read-only dictionary of inherited providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: providers
Read-only dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: overridden_by
Tuple of overriding catalogs.
:type: tuple[:py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`]
.. py:attribute:: is_overridden
Read-only property that is set to ``True`` if catalog is overridden.
:type: bool
.. py:attribute:: is_overridden
Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
.. py:attribute:: provider_type
If provider type is defined, :py:class:`DeclarativeCatalog` checks that
all of its providers are instances of
:py:attr:`DeclarativeCatalog.provider_type`.
:type: type | None
"""
Bundle = CatalogBundle
name = str()
cls_providers = dict()
inherited_providers = dict()
providers = dict()
overridden_by = tuple()
is_overridden = bool
last_overriding = None
provider_type = None
_catalog = DynamicCatalog
__IS_CATALOG__ = True
@classmethod
def is_bundle_owner(cls, bundle):
"""Check if catalog is bundle owner.
:param bundle: Catalog's bundle instance.
:type bundle: :py:class:`CatalogBundle`
:rtype: bool
"""
return cls._catalog.is_bundle_owner(bundle)
@classmethod
def get_provider_bind_name(cls, provider):
"""Return provider's name in catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider's name.
:rtype: str
"""
return cls._catalog.get_provider_bind_name(provider)
@classmethod
def is_provider_bound(cls, provider):
"""Check if provider is bound to the catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:rtype: bool
"""
return cls._catalog.is_provider_bound(provider)
@classmethod
def filter(cls, provider_type):
"""Return dictionary of providers, that are instance of provided type.
:param provider_type: Provider's type.
:type provider_type: :py:class:`dependency_injector.providers.Provider`
:rtype: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return cls._catalog.filter(provider_type)
@classmethod
def override(cls, overriding):
"""Override current catalog providers by overriding catalog providers.
:param overriding: Overriding catalog.
:type overriding: :py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`
:raise: :py:exc:`dependency_injector.errors.Error` if trying to
override catalog by itself or its subclasses
:rtype: None
"""
if is_declarative_catalog(overriding) and issubclass(cls, overriding):
raise Error('Catalog {0} could not be overridden '
'with itself or its subclasses'.format(cls))
return cls._catalog.override(overriding)
@classmethod
def reset_last_overriding(cls):
"""Reset last overriding catalog.
:rtype: None
"""
cls._catalog.reset_last_overriding()
@classmethod
def reset_override(cls):
"""Reset all overridings for all catalog providers.
:rtype: None
"""
cls._catalog.reset_override()
@classmethod
def get_provider(cls, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
return cls._catalog.get_provider(name)
get = get_provider # Backward compatibility for versions < 0.11.*
@classmethod
def bind_provider(cls, name, provider):
"""Bind provider to catalog with specified name.
:param name: Name of the provider.
:type name: str
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
setattr(cls, name, provider)
@classmethod
def bind_providers(cls, providers):
"""Bind providers dictionary to catalog.
:param providers: Dictionary of providers, where key is a name
and value is a provider.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
for name, provider in six.iteritems(providers):
setattr(cls, name, provider)
@classmethod
def has_provider(cls, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return hasattr(cls, name)
has = has_provider # Backward compatibility for versions < 0.11.*
@classmethod
def unbind_provider(cls, name):
"""Remove provider binding.
:param name: Provider's name.
:type name: str
:rtype: None
"""
delattr(cls, name)
@classmethod
def __getattr__(cls, name): # pragma: no cover
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
raise NotImplementedError('Implementated in metaclass')
@classmethod
def __setattr__(cls, name, value): # pragma: no cover
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
raise NotImplementedError('Implementated in metaclass')
@classmethod
def __delattr__(cls, name): # pragma: no cover
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
raise NotImplementedError('Implementated in metaclass')
# Backward compatibility for versions < 0.11.*
AbstractCatalog = DeclarativeCatalog

333
dependency_injector/catalogs/dynamic.py Normal file
View File

@ -0,0 +1,333 @@
"""Dependency injector dynamic catalog module."""
import six
from dependency_injector.catalogs.bundle import CatalogBundle
from dependency_injector.utils import (
is_provider,
ensure_is_provider,
ensure_is_catalog_bundle,
)
from dependency_injector.errors import (
Error,
UndefinedProviderError,
)
@six.python_2_unicode_compatible
class DynamicCatalog(object):
"""Dynamic catalog of providers.
:py:class:`DynamicCatalog` is a catalog of providers that could be created
in application's runtime. It should cover most of the cases when list of
providers that would be included in catalog is non-deterministic in terms
of apllication code (catalog's structure could be determined just after
application will be started and will do some initial work, like parsing
list of catalog's providers from the configuration).
.. code-block:: python
services = DynamicCatalog(auth=providers.Factory(AuthService),
users=providers.Factory(UsersService))
users_service = services.users()
.. py:attribute:: Bundle
Catalog's bundle class.
:type: :py:class:`CatalogBundle`
.. py:attribute:: name
Catalog's name.
By default, it is catalog's module + catalog's class name.
:type: str
.. py:attribute:: providers
Dictionary of all providers.
:type: dict[str, :py:class:`dependency_injector.providers.Provider`]
.. py:attribute:: overridden_by
Tuple of overriding catalogs.
:type: tuple[
:py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog`]
.. py:attribute:: provider_type
If provider type is defined, :py:class:`DynamicCatalog` checks that
all of its providers are instances of
:py:attr:`DynamicCatalog.provider_type`.
:type: type | None
"""
provider_type = None
__IS_CATALOG__ = True
__slots__ = ('name', 'providers', 'provider_names', 'overridden_by',
'Bundle')
def __init__(self, **providers):
"""Initializer.
:param providers: Dictionary of catalog providers.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
self.Bundle = CatalogBundle.sub_cls_factory(self)
self.name = '.'.join((self.__class__.__module__,
self.__class__.__name__))
self.providers = dict()
self.provider_names = dict()
self.overridden_by = tuple()
self.bind_providers(providers)
super(DynamicCatalog, self).__init__()
def is_bundle_owner(self, bundle):
"""Check if catalog is bundle owner.
:param bundle: Catalog's bundle instance.
:type bundle: :py:class:`CatalogBundle`
:rtype: bool
"""
return ensure_is_catalog_bundle(bundle) and bundle.catalog is self
def get_provider_bind_name(self, provider):
"""Return provider's name in catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider's name.
:rtype: str
"""
if not self.is_provider_bound(provider):
raise Error('Can not find bind name for {0} in catalog {1}'.format(
provider, self))
return self.provider_names[provider]
def is_provider_bound(self, provider):
"""Check if provider is bound to the catalog.
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:rtype: bool
"""
return provider in self.provider_names
def filter(self, provider_type):
"""Return dictionary of providers, that are instance of provided type.
:param provider_type: Provider's type.
:type provider_type: :py:class:`dependency_injector.providers.Provider`
:rtype: dict[str, :py:class:`dependency_injector.providers.Provider`]
"""
return dict((name, provider)
for name, provider in six.iteritems(self.providers)
if isinstance(provider, provider_type))
@property
def is_overridden(self):
"""Read-only property that is set to ``True`` if catalog is overridden.
:rtype: bool
"""
return bool(self.overridden_by)
@property
def last_overriding(self):
"""Read-only reference to the last overriding catalog, if any.
:type: :py:class:`DeclarativeCatalog` | :py:class:`DynamicCatalog` |
None
"""
return self.overridden_by[-1] if self.overridden_by else None
def override(self, overriding):
"""Override current catalog providers by overriding catalog providers.
:param overriding: Overriding catalog.
:type overriding: :py:class:`DeclarativeCatalog` |
:py:class:`DynamicCatalog`
:raise: :py:exc:`dependency_injector.errors.Error` if trying to
override catalog by itself
:rtype: None
"""
if overriding is self:
raise Error('Catalog {0} could not be overridden '
'with itself'.format(self))
self.overridden_by += (overriding,)
for name, provider in six.iteritems(overriding.providers):
self.get_provider(name).override(provider)
def reset_last_overriding(self):
"""Reset last overriding catalog.
:rtype: None
"""
if not self.is_overridden:
raise Error('Catalog {0} is not overridden'.format(self))
self.overridden_by = self.overridden_by[:-1]
for provider in six.itervalues(self.providers):
provider.reset_last_overriding()
def reset_override(self):
"""Reset all overridings for all catalog providers.
:rtype: None
"""
self.overridden_by = tuple()
for provider in six.itervalues(self.providers):
provider.reset_override()
def get_provider(self, name):
"""Return provider with specified name or raise an error.
:param name: Provider's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
:return: Provider with specified name.
:rtype: :py:class:`dependency_injector.providers.Provider`
"""
try:
return self.providers[name]
except KeyError:
raise UndefinedProviderError('{0} has no provider with such '
'name - {1}'.format(self, name))
def bind_provider(self, name, provider):
"""Bind provider to catalog with specified name.
:param name: Name of the provider.
:type name: str
:param provider: Provider instance.
:type provider: :py:class:`dependency_injector.providers.Provider`
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
provider = ensure_is_provider(provider)
if (self.__class__.provider_type and
not isinstance(provider, self.__class__.provider_type)):
raise Error('{0} can contain only {1} instances'.format(
self, self.__class__.provider_type))
if name in self.providers:
raise Error('Catalog {0} already has provider with '
'such name - {1}'.format(self, name))
if provider in self.provider_names:
raise Error('Catalog {0} already has such provider '
'instance - {1}'.format(self, provider))
self.providers[name] = provider
self.provider_names[provider] = name
def bind_providers(self, providers):
"""Bind providers dictionary to catalog.
:param providers: Dictionary of providers, where key is a name
and value is a provider.
:type providers:
dict[str, :py:class:`dependency_injector.providers.Provider`]
:raise: :py:exc:`dependency_injector.errors.Error`
:rtype: None
"""
for name, provider in six.iteritems(providers):
self.bind_provider(name, provider)
def has_provider(self, name):
"""Check if there is provider with certain name.
:param name: Provider's name.
:type name: str
:rtype: bool
"""
return name in self.providers
def unbind_provider(self, name):
"""Remove provider binding.
:param name: Provider's name.
:type name: str
:rtype: None
"""
provider = self.get_provider(name)
del self.providers[name]
del self.provider_names[provider]
def __getattr__(self, name):
"""Return provider with specified name or raise en error.
:param name: Attribute's name.
:type name: str
:raise: :py:exc:`dependency_injector.errors.UndefinedProviderError`
"""
return self.get_provider(name)
def __setattr__(self, name, value):
"""Handle setting of catalog attributes.
Setting of attributes works as usual, but if value of attribute is
provider, this provider will be bound to catalog.
:param name: Attribute's name.
:type name: str
:param value: Attribute's value.
:type value: :py:class:`dependency_injector.providers.Provider` |
object
:rtype: None
"""
if is_provider(value):
return self.bind_provider(name, value)
return super(DynamicCatalog, self).__setattr__(name, value)
def __delattr__(self, name):
"""Handle deleting of catalog attibute.
Deleting of attributes works as usual, but if value of attribute is
provider, this provider will be unbound from catalog.
:param name: Attribute's name.
:type name: str
:rtype: None
"""
self.unbind_provider(name)
def __repr__(self):
"""Return Python representation of catalog.
:rtype: str
"""
return '<{0}({1})>'.format(self.name,
', '.join(six.iterkeys(self.providers)))
__str__ = __repr__

4
docs/main/changelog.rst
View File

@ -9,8 +9,10 @@ follows `Semantic versioning`_
Development version Development version
------------------- -------------------
- [Refactoring] split ``catalogs`` module into smaller modules,
``catalogs`` module become a package.
- [Refactoring] split ``providers`` module into smaller modules, - [Refactoring] split ``providers`` module into smaller modules,
``providers`` become a package. ``providers`` module become a package.
1.15.1 1.15.1
------ ------

1
setup.py
View File

@ -53,6 +53,7 @@ setup(name='dependency_injector',
download_url='https://pypi.python.org/pypi/dependency_injector', download_url='https://pypi.python.org/pypi/dependency_injector',
license='BSD New', license='BSD New',
packages=['dependency_injector', packages=['dependency_injector',
'dependency_injector.catalogs',
'dependency_injector.providers'], 'dependency_injector.providers'],
platforms=['any'], platforms=['any'],
zip_safe=True, zip_safe=True,

1
tests/catalogs/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Dependency injector catalogs unittests."""

101
tests/catalogs/test_bundle.py Normal file
View File

@ -0,0 +1,101 @@
"""Dependency injector catalog bundles unittests."""
import unittest2 as unittest
from dependency_injector import (
catalogs,
providers,
errors,
)
class CatalogA(catalogs.DeclarativeCatalog):
"""Test catalog A."""
p11 = providers.Provider()
p12 = providers.Provider()
class CatalogB(CatalogA):
"""Test catalog B."""
p21 = providers.Provider()
p22 = providers.Provider()
class CatalogBundleTests(unittest.TestCase):
"""Catalog bundle test cases."""
def setUp(self):
"""Set test environment up."""
self.bundle = CatalogB.Bundle(CatalogB.p11,
CatalogB.p12)
def test_get_attr_from_bundle(self):
"""Test get providers (attribute) from catalog bundle."""
self.assertIs(self.bundle.p11, CatalogA.p11)
self.assertIs(self.bundle.p12, CatalogA.p12)
def test_get_attr_not_from_bundle(self):
"""Test get providers (attribute) that are not in bundle."""
self.assertRaises(errors.Error, getattr, self.bundle, 'p21')
self.assertRaises(errors.Error, getattr, self.bundle, 'p22')
def test_get_method_from_bundle(self):
"""Test get providers (get() method) from bundle."""
self.assertIs(self.bundle.get_provider('p11'), CatalogB.p11)
self.assertIs(self.bundle.get_provider('p12'), CatalogB.p12)
def test_get_method_not_from_bundle(self):
"""Test get providers (get() method) that are not in bundle."""
self.assertRaises(errors.Error, self.bundle.get_provider, 'p21')
self.assertRaises(errors.Error, self.bundle.get_provider, 'p22')
def test_has(self):
"""Test checks of providers availability in bundle."""
self.assertTrue(self.bundle.has_provider('p11'))
self.assertTrue(self.bundle.has_provider('p12'))
self.assertFalse(self.bundle.has_provider('p21'))
self.assertFalse(self.bundle.has_provider('p22'))
def test_hasattr(self):
"""Test checks of providers availability in bundle."""
self.assertTrue(hasattr(self.bundle, 'p11'))
self.assertTrue(hasattr(self.bundle, 'p12'))
self.assertFalse(hasattr(self.bundle, 'p21'))
self.assertFalse(hasattr(self.bundle, 'p22'))
def test_create_bundle_with_unbound_provider(self):
"""Test that bundle is not created with unbound provider."""
self.assertRaises(errors.Error, CatalogB.Bundle, providers.Provider())
def test_create_bundle_with_another_catalog_provider(self):
"""Test that bundle can not contain another catalog's provider."""
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
provider = providers.Provider()
self.assertRaises(errors.Error,
CatalogB.Bundle, CatalogB.p21, TestCatalog.provider)
def test_create_bundle_with_another_catalog_provider_with_same_name(self):
"""Test that bundle can not contain another catalog's provider."""
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
p21 = providers.Provider()
self.assertRaises(errors.Error,
CatalogB.Bundle, CatalogB.p21, TestCatalog.p21)
def test_is_bundle_owner(self):
"""Test that catalog bundle is owned by catalog."""
self.assertTrue(CatalogB.is_bundle_owner(self.bundle))
self.assertFalse(CatalogA.is_bundle_owner(self.bundle))
def test_is_bundle_owner_with_not_bundle_instance(self):
"""Test that check of bundle ownership raises error with not bundle."""
self.assertRaises(errors.Error, CatalogB.is_bundle_owner, object())

14
tests/catalogs/test_common.py Normal file
View File

@ -0,0 +1,14 @@
"""Dependency injector common catalogs unittests."""
import unittest2 as unittest
class CatalogModuleBackwardCompatibility(unittest.TestCase):
"""Backward compatibility test of catalog module."""
def test_import_catalog(self):
"""Test that module `catalog` is the same as `catalogs`."""
from dependency_injector import catalog
from dependency_injector import catalogs
self.assertIs(catalog, catalogs)

352
tests/catalogs/test_declarative.py Normal file
View File

@ -0,0 +1,352 @@
"""Dependency injector declarative catalog unittests."""
import unittest2 as unittest
from dependency_injector import (
catalogs,
providers,
injections,
errors,
)
class CatalogA(catalogs.DeclarativeCatalog):
"""Test catalog A."""
p11 = providers.Provider()
p12 = providers.Provider()
class CatalogB(CatalogA):
"""Test catalog B."""
p21 = providers.Provider()
p22 = providers.Provider()
class DeclarativeCatalogTests(unittest.TestCase):
"""Declarative catalog tests."""
def test_cls_providers(self):
"""Test `di.DeclarativeCatalog.cls_providers` contents."""
self.assertDictEqual(CatalogA.cls_providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
self.assertDictEqual(CatalogB.cls_providers,
dict(p21=CatalogB.p21,
p22=CatalogB.p22))
def test_inherited_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(CatalogA.inherited_providers, dict())
self.assertDictEqual(CatalogB.inherited_providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
def test_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(CatalogA.providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
self.assertDictEqual(CatalogB.providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12,
p21=CatalogB.p21,
p22=CatalogB.p22))
def test_bind_provider(self):
"""Test setting of provider via bind_provider() to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogA.bind_provider('px', px)
CatalogA.bind_provider('py', py)
self.assertIs(CatalogA.px, px)
self.assertIs(CatalogA.get_provider('px'), px)
self.assertIs(CatalogA.py, py)
self.assertIs(CatalogA.get_provider('py'), py)
del CatalogA.px
del CatalogA.py
def test_bind_provider_with_valid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DeclarativeCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = SomeProvider()
py = SomeProvider()
SomeCatalog.bind_provider('px', px)
SomeCatalog.py = py
self.assertIs(SomeCatalog.px, px)
self.assertIs(SomeCatalog.get_provider('px'), px)
self.assertIs(SomeCatalog.py, py)
self.assertIs(SomeCatalog.get_provider('py'), py)
def test_bind_provider_with_invalid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DeclarativeCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = providers.Provider()
with self.assertRaises(errors.Error):
SomeCatalog.bind_provider('px', px)
with self.assertRaises(errors.Error):
SomeCatalog.px = px
with self.assertRaises(errors.Error):
SomeCatalog.bind_providers(dict(px=px))
def test_bind_providers(self):
"""Test setting of provider via bind_providers() to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogB.bind_providers(dict(px=px, py=py))
self.assertIs(CatalogB.px, px)
self.assertIs(CatalogB.get_provider('px'), px)
self.assertIs(CatalogB.py, py)
self.assertIs(CatalogB.get_provider('py'), py)
del CatalogB.px
del CatalogB.py
def test_setattr(self):
"""Test setting of providers via attributes to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogB.px = px
CatalogB.py = py
self.assertIs(CatalogB.px, px)
self.assertIs(CatalogB.get_provider('px'), px)
self.assertIs(CatalogB.py, py)
self.assertIs(CatalogB.get_provider('py'), py)
del CatalogB.px
del CatalogB.py
def test_unbind_provider(self):
"""Test that catalog unbinds provider correct."""
CatalogB.px = providers.Provider()
CatalogB.unbind_provider('px')
self.assertFalse(CatalogB.has_provider('px'))
def test_unbind_via_delattr(self):
"""Test that catalog unbinds provider correct."""
CatalogB.px = providers.Provider()
del CatalogB.px
self.assertFalse(CatalogB.has_provider('px'))
def test_provider_is_bound(self):
"""Test that providers are bound to the catalogs."""
self.assertTrue(CatalogA.is_provider_bound(CatalogA.p11))
self.assertEquals(CatalogA.get_provider_bind_name(CatalogA.p11), 'p11')
self.assertTrue(CatalogA.is_provider_bound(CatalogA.p12))
self.assertEquals(CatalogA.get_provider_bind_name(CatalogA.p12), 'p12')
def test_provider_binding_to_different_catalogs(self):
"""Test that provider could be bound to different catalogs."""
p11 = CatalogA.p11
p12 = CatalogA.p12
class CatalogD(catalogs.DeclarativeCatalog):
"""Test catalog."""
pd1 = p11
pd2 = p12
class CatalogE(catalogs.DeclarativeCatalog):
"""Test catalog."""
pe1 = p11
pe2 = p12
self.assertTrue(CatalogA.is_provider_bound(p11))
self.assertTrue(CatalogD.is_provider_bound(p11))
self.assertTrue(CatalogE.is_provider_bound(p11))
self.assertEquals(CatalogA.get_provider_bind_name(p11), 'p11')
self.assertEquals(CatalogD.get_provider_bind_name(p11), 'pd1')
self.assertEquals(CatalogE.get_provider_bind_name(p11), 'pe1')
self.assertTrue(CatalogA.is_provider_bound(p12))
self.assertTrue(CatalogD.is_provider_bound(p12))
self.assertTrue(CatalogE.is_provider_bound(p12))
self.assertEquals(CatalogA.get_provider_bind_name(p12), 'p12')
self.assertEquals(CatalogD.get_provider_bind_name(p12), 'pd2')
self.assertEquals(CatalogE.get_provider_bind_name(p12), 'pe2')
def test_provider_rebinding_to_the_same_catalog(self):
"""Test provider rebinding to the same catalog."""
with self.assertRaises(errors.Error):
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
p1 = providers.Provider()
p2 = p1
def test_provider_rebinding_to_the_same_catalogs_hierarchy(self):
"""Test provider rebinding to the same catalogs hierarchy."""
class TestCatalog1(catalogs.DeclarativeCatalog):
"""Test catalog."""
p1 = providers.Provider()
with self.assertRaises(errors.Error):
class TestCatalog2(TestCatalog1):
"""Test catalog."""
p2 = TestCatalog1.p1
def test_get(self):
"""Test getting of providers using get() method."""
self.assertIs(CatalogB.get('p11'), CatalogB.p11)
self.assertIs(CatalogB.get('p12'), CatalogB.p12)
self.assertIs(CatalogB.get('p22'), CatalogB.p22)
self.assertIs(CatalogB.get('p22'), CatalogB.p22)
self.assertIs(CatalogB.get_provider('p11'), CatalogB.p11)
self.assertIs(CatalogB.get_provider('p12'), CatalogB.p12)
self.assertIs(CatalogB.get_provider('p22'), CatalogB.p22)
self.assertIs(CatalogB.get_provider('p22'), CatalogB.p22)
def test_get_undefined(self):
"""Test getting of undefined providers using get() method."""
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.get('undefined')
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.get_provider('undefined')
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.undefined
def test_has(self):
"""Test checks of providers availability in catalog."""
self.assertTrue(CatalogB.has('p11'))
self.assertTrue(CatalogB.has('p12'))
self.assertTrue(CatalogB.has('p21'))
self.assertTrue(CatalogB.has('p22'))
self.assertFalse(CatalogB.has('undefined'))
self.assertTrue(CatalogB.has_provider('p11'))
self.assertTrue(CatalogB.has_provider('p12'))
self.assertTrue(CatalogB.has_provider('p21'))
self.assertTrue(CatalogB.has_provider('p22'))
self.assertFalse(CatalogB.has_provider('undefined'))
def test_filter_all_providers_by_type(self):
"""Test getting of all catalog providers of specific type."""
self.assertTrue(len(CatalogB.filter(providers.Provider)) == 4)
self.assertTrue(len(CatalogB.filter(providers.Value)) == 0)
def test_repr(self):
"""Test catalog representation."""
self.assertIn('CatalogA', repr(CatalogA))
self.assertIn('p11', repr(CatalogA))
self.assertIn('p12', repr(CatalogA))
self.assertIn('CatalogB', repr(CatalogB))
self.assertIn('p11', repr(CatalogB))
self.assertIn('p12', repr(CatalogB))
self.assertIn('p21', repr(CatalogB))
self.assertIn('p22', repr(CatalogB))
def test_abstract_catalog_backward_compatibility(self):
"""Test that di.AbstractCatalog is available."""
self.assertIs(catalogs.DeclarativeCatalog, catalogs.AbstractCatalog)
class TestCatalogWithProvidingCallbacks(unittest.TestCase):
"""Catalog with providing callback tests."""
def test_concept(self):
"""Test concept."""
class UsersService(object):
"""Users service, that has dependency on database."""
class AuthService(object):
"""Auth service, that has dependencies on users service."""
def __init__(self, users_service):
"""Initializer."""
self.users_service = users_service
class Services(catalogs.DeclarativeCatalog):
"""Catalog of service providers."""
@providers.Factory
def users():
"""Provide users service.
:rtype: providers.Provider -> UsersService
"""
return UsersService()
@providers.Factory
@injections.inject(users_service=users)
def auth(**kwargs):
"""Provide users service.
:rtype: providers.Provider -> AuthService
"""
return AuthService(**kwargs)
# Retrieving catalog providers:
users_service = Services.users()
auth_service = Services.auth()
# Making some asserts:
self.assertIsInstance(auth_service.users_service, UsersService)
self.assertIsNot(users_service, Services.users())
self.assertIsNot(auth_service, Services.auth())
# Overriding auth service provider and making some asserts:
class ExtendedAuthService(AuthService):
"""Extended version of auth service."""
def __init__(self, users_service, ttl):
"""Initializer."""
self.ttl = ttl
super(ExtendedAuthService, self).__init__(
users_service=users_service)
class OverriddenServices(Services):
"""Catalog of service providers."""
@providers.override(Services.auth)
@providers.Factory
@injections.inject(users_service=Services.users)
@injections.inject(ttl=3600)
def auth(**kwargs):
"""Provide users service.
:rtype: providers.Provider -> AuthService
"""
return ExtendedAuthService(**kwargs)
auth_service = Services.auth()
self.assertIsInstance(auth_service, ExtendedAuthService)

197
tests/catalogs/test_dynamic.py Normal file
View File

@ -0,0 +1,197 @@
"""Dependency injector dynamic catalog unittests."""
import unittest2 as unittest
from dependency_injector import (
catalogs,
providers,
errors,
)
class DynamicCatalogTests(unittest.TestCase):
"""Dynamic catalog tests."""
catalog = None
""":type: di.DynamicCatalog"""
def setUp(self):
"""Set test environment up."""
self.catalog = catalogs.DynamicCatalog(p1=providers.Provider(),
p2=providers.Provider())
self.catalog.name = 'TestCatalog'
def test_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(self.catalog.providers,
dict(p1=self.catalog.p1,
p2=self.catalog.p2))
def test_bind_provider(self):
"""Test setting of provider via bind_provider() to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.bind_provider('px', px)
self.catalog.bind_provider('py', py)
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_bind_provider_with_valid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DynamicCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = SomeProvider()
py = SomeProvider()
catalog = SomeCatalog()
catalog.bind_provider('px', px)
catalog.py = py
self.assertIs(catalog.px, px)
self.assertIs(catalog.get_provider('px'), px)
self.assertIs(catalog.py, py)
self.assertIs(catalog.get_provider('py'), py)
def test_bind_provider_with_invalid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DynamicCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = providers.Provider()
catalog = SomeCatalog()
with self.assertRaises(errors.Error):
catalog.bind_provider('px', px)
with self.assertRaises(errors.Error):
catalog.px = px
with self.assertRaises(errors.Error):
catalog.bind_providers(dict(px=px))
def test_bind_providers(self):
"""Test setting of provider via bind_providers() to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.bind_providers(dict(px=px, py=py))
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_setattr(self):
"""Test setting of providers via attributes to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.px = px
self.catalog.py = py
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_unbind_provider(self):
"""Test that catalog unbinds provider correct."""
self.catalog.px = providers.Provider()
self.catalog.unbind_provider('px')
self.assertFalse(self.catalog.has_provider('px'))
def test_unbind_via_delattr(self):
"""Test that catalog unbinds provider correct."""
self.catalog.px = providers.Provider()
del self.catalog.px
self.assertFalse(self.catalog.has_provider('px'))
def test_provider_is_bound(self):
"""Test that providers are bound to the catalogs."""
self.assertTrue(self.catalog.is_provider_bound(self.catalog.p1))
self.assertEquals(
self.catalog.get_provider_bind_name(self.catalog.p1), 'p1')
self.assertTrue(self.catalog.is_provider_bound(self.catalog.p2))
self.assertEquals(
self.catalog.get_provider_bind_name(self.catalog.p2), 'p2')
def test_provider_binding_to_different_catalogs(self):
"""Test that provider could be bound to different catalogs."""
p1 = self.catalog.p1
p2 = self.catalog.p2
catalog_a = catalogs.DynamicCatalog(pa1=p1, pa2=p2)
catalog_b = catalogs.DynamicCatalog(pb1=p1, pb2=p2)
self.assertTrue(self.catalog.is_provider_bound(p1))
self.assertTrue(catalog_a.is_provider_bound(p1))
self.assertTrue(catalog_b.is_provider_bound(p1))
self.assertEquals(self.catalog.get_provider_bind_name(p1), 'p1')
self.assertEquals(catalog_a.get_provider_bind_name(p1), 'pa1')
self.assertEquals(catalog_b.get_provider_bind_name(p1), 'pb1')
self.assertTrue(self.catalog.is_provider_bound(p2))
self.assertTrue(catalog_a.is_provider_bound(p2))
self.assertTrue(catalog_b.is_provider_bound(p2))
self.assertEquals(self.catalog.get_provider_bind_name(p2), 'p2')
self.assertEquals(catalog_a.get_provider_bind_name(p2), 'pa2')
self.assertEquals(catalog_b.get_provider_bind_name(p2), 'pb2')
def test_provider_rebinding_to_the_same_catalog(self):
"""Test provider rebinding to the same catalog."""
with self.assertRaises(errors.Error):
self.catalog.p3 = self.catalog.p1
def test_provider_binding_with_the_same_name(self):
"""Test binding of provider with the same name."""
with self.assertRaises(errors.Error):
self.catalog.bind_provider('p1', providers.Provider())
def test_get(self):
"""Test getting of providers using get() method."""
self.assertIs(self.catalog.get_provider('p1'), self.catalog.p1)
self.assertIs(self.catalog.get_provider('p2'), self.catalog.p2)
def test_get_undefined(self):
"""Test getting of undefined providers using get() method."""
with self.assertRaises(errors.UndefinedProviderError):
self.catalog.get_provider('undefined')
with self.assertRaises(errors.UndefinedProviderError):
self.catalog.undefined
def test_has_provider(self):
"""Test checks of providers availability in catalog."""
self.assertTrue(self.catalog.has_provider('p1'))
self.assertTrue(self.catalog.has_provider('p2'))
self.assertFalse(self.catalog.has_provider('undefined'))
def test_filter_all_providers_by_type(self):
"""Test getting of all catalog providers of specific type."""
self.assertTrue(len(self.catalog.filter(providers.Provider)) == 2)
self.assertTrue(len(self.catalog.filter(providers.Value)) == 0)
def test_repr(self):
"""Test catalog representation."""
self.assertIn('TestCatalog', repr(self.catalog))
self.assertIn('p1', repr(self.catalog))
self.assertIn('p2', repr(self.catalog))

142
tests/catalogs/test_override.py Normal file
View File

@ -0,0 +1,142 @@
"""Dependency injector catalogs overriding unittests."""
import unittest2 as unittest
from dependency_injector import (
catalogs,
providers,
errors,
)
class CatalogA(catalogs.DeclarativeCatalog):
"""Test catalog A."""
p11 = providers.Provider()
p12 = providers.Provider()
class CatalogB(CatalogA):
"""Test catalog B."""
p21 = providers.Provider()
p22 = providers.Provider()
class OverrideTests(unittest.TestCase):
"""Catalog overriding and override decorator test cases."""
def tearDown(self):
"""Tear test environment down."""
CatalogA.reset_override()
def test_overriding(self):
"""Test catalog overriding with another catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
self.assertEqual(len(CatalogA.overridden_by), 1)
def test_override_declarative_catalog_with_itself(self):
"""Test catalog overriding of declarative catalog with itself."""
with self.assertRaises(errors.Error):
CatalogA.override(CatalogA)
def test_override_declarative_catalog_with_subclass(self):
"""Test catalog overriding of declarative catalog with subclass."""
with self.assertRaises(errors.Error):
CatalogB.override(CatalogA)
def test_override_dynamic_catalog_with_itself(self):
"""Test catalog overriding of dynamic catalog with itself."""
catalog = catalogs.DynamicCatalog(p11=providers.Value(1),
p12=providers.Value(2))
with self.assertRaises(errors.Error):
catalog.override(catalog)
def test_overriding_with_dynamic_catalog(self):
"""Test catalog overriding with another dynamic catalog."""
CatalogA.override(catalogs.DynamicCatalog(p11=providers.Value(1),
p12=providers.Value(2)))
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
self.assertEqual(len(CatalogA.overridden_by), 1)
def test_is_overridden(self):
"""Test catalog is_overridden property."""
self.assertFalse(CatalogA.is_overridden)
@catalogs.override(CatalogA)
class OverridingCatalog(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
self.assertTrue(CatalogA.is_overridden)
def test_last_overriding(self):
"""Test catalog last_overriding property."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
self.assertIs(CatalogA.last_overriding, OverridingCatalog2)
def test_last_overriding_on_not_overridden(self):
"""Test catalog last_overriding property on not overridden catalog."""
self.assertIsNone(CatalogA.last_overriding)
def test_reset_last_overriding(self):
"""Test resetting last overriding catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(3)
p12 = providers.Value(4)
CatalogA.reset_last_overriding()
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
def test_reset_last_overriding_when_not_overridden(self):
"""Test resetting last overriding catalog when it is not overridden."""
with self.assertRaises(errors.Error):
CatalogA.reset_last_overriding()
def test_reset_override(self):
"""Test resetting all catalog overrides."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(3)
p12 = providers.Value(4)
CatalogA.reset_override()
self.assertFalse(CatalogA.p11.is_overridden)
self.assertFalse(CatalogA.p12.is_overridden)

8
tests/providers/test_base.py
View File

@ -2,9 +2,11 @@
import unittest2 as unittest import unittest2 as unittest
from dependency_injector import providers from dependency_injector import (
from dependency_injector import errors providers,
from dependency_injector import utils utils,
errors,
)
class ProviderTests(unittest.TestCase): class ProviderTests(unittest.TestCase):

10
tests/providers/test_callable.py
View File

@ -2,10 +2,12 @@
import unittest2 as unittest import unittest2 as unittest
from dependency_injector import providers from dependency_injector import (
from dependency_injector import injections providers,
from dependency_injector import utils injections,
from dependency_injector import errors utils,
errors,
)
class CallableTests(unittest.TestCase): class CallableTests(unittest.TestCase):

8
tests/providers/test_config.py
View File

@ -2,9 +2,11 @@
import unittest2 as unittest import unittest2 as unittest
from dependency_injector import providers from dependency_injector import (
from dependency_injector import utils providers,
from dependency_injector import errors utils,
errors,
)
class ConfigTests(unittest.TestCase): class ConfigTests(unittest.TestCase):

10
tests/providers/test_creational.py
View File

@ -2,10 +2,12 @@
import unittest2 as unittest import unittest2 as unittest
from dependency_injector import providers from dependency_injector import (
from dependency_injector import injections providers,
from dependency_injector import utils injections,
from dependency_injector import errors utils,
errors,
)
class Example(object): class Example(object):

6
tests/providers/test_static.py
View File

@ -2,8 +2,10 @@
import unittest2 as unittest import unittest2 as unittest
from dependency_injector import providers from dependency_injector import (
from dependency_injector import utils providers,
utils,
)
class StaticProvidersTests(unittest.TestCase): class StaticProvidersTests(unittest.TestCase):

746
tests/test_catalogs.py
View File

@ -1,746 +0,0 @@
"""Dependency injector catalogs unittests."""
import unittest2 as unittest
from dependency_injector import catalogs
from dependency_injector import providers
from dependency_injector import injections
from dependency_injector import errors
class CatalogA(catalogs.DeclarativeCatalog):
"""Test catalog A."""
p11 = providers.Provider()
p12 = providers.Provider()
class CatalogB(CatalogA):
"""Test catalog B."""
p21 = providers.Provider()
p22 = providers.Provider()
class CatalogBundleTests(unittest.TestCase):
"""Catalog bundle test cases."""
def setUp(self):
"""Set test environment up."""
self.bundle = CatalogB.Bundle(CatalogB.p11,
CatalogB.p12)
def test_get_attr_from_bundle(self):
"""Test get providers (attribute) from catalog bundle."""
self.assertIs(self.bundle.p11, CatalogA.p11)
self.assertIs(self.bundle.p12, CatalogA.p12)
def test_get_attr_not_from_bundle(self):
"""Test get providers (attribute) that are not in bundle."""
self.assertRaises(errors.Error, getattr, self.bundle, 'p21')
self.assertRaises(errors.Error, getattr, self.bundle, 'p22')
def test_get_method_from_bundle(self):
"""Test get providers (get() method) from bundle."""
self.assertIs(self.bundle.get_provider('p11'), CatalogB.p11)
self.assertIs(self.bundle.get_provider('p12'), CatalogB.p12)
def test_get_method_not_from_bundle(self):
"""Test get providers (get() method) that are not in bundle."""
self.assertRaises(errors.Error, self.bundle.get_provider, 'p21')
self.assertRaises(errors.Error, self.bundle.get_provider, 'p22')
def test_has(self):
"""Test checks of providers availability in bundle."""
self.assertTrue(self.bundle.has_provider('p11'))
self.assertTrue(self.bundle.has_provider('p12'))
self.assertFalse(self.bundle.has_provider('p21'))
self.assertFalse(self.bundle.has_provider('p22'))
def test_hasattr(self):
"""Test checks of providers availability in bundle."""
self.assertTrue(hasattr(self.bundle, 'p11'))
self.assertTrue(hasattr(self.bundle, 'p12'))
self.assertFalse(hasattr(self.bundle, 'p21'))
self.assertFalse(hasattr(self.bundle, 'p22'))
def test_create_bundle_with_unbound_provider(self):
"""Test that bundle is not created with unbound provider."""
self.assertRaises(errors.Error, CatalogB.Bundle, providers.Provider())
def test_create_bundle_with_another_catalog_provider(self):
"""Test that bundle can not contain another catalog's provider."""
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
provider = providers.Provider()
self.assertRaises(errors.Error,
CatalogB.Bundle, CatalogB.p21, TestCatalog.provider)
def test_create_bundle_with_another_catalog_provider_with_same_name(self):
"""Test that bundle can not contain another catalog's provider."""
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
p21 = providers.Provider()
self.assertRaises(errors.Error,
CatalogB.Bundle, CatalogB.p21, TestCatalog.p21)
def test_is_bundle_owner(self):
"""Test that catalog bundle is owned by catalog."""
self.assertTrue(CatalogB.is_bundle_owner(self.bundle))
self.assertFalse(CatalogA.is_bundle_owner(self.bundle))
def test_is_bundle_owner_with_not_bundle_instance(self):
"""Test that check of bundle ownership raises error with not bundle."""
self.assertRaises(errors.Error, CatalogB.is_bundle_owner, object())
class DynamicCatalogTests(unittest.TestCase):
"""Dynamic catalog tests."""
catalog = None
""":type: di.DynamicCatalog"""
def setUp(self):
"""Set test environment up."""
self.catalog = catalogs.DynamicCatalog(p1=providers.Provider(),
p2=providers.Provider())
self.catalog.name = 'TestCatalog'
def test_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(self.catalog.providers,
dict(p1=self.catalog.p1,
p2=self.catalog.p2))
def test_bind_provider(self):
"""Test setting of provider via bind_provider() to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.bind_provider('px', px)
self.catalog.bind_provider('py', py)
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_bind_provider_with_valid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DynamicCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = SomeProvider()
py = SomeProvider()
catalog = SomeCatalog()
catalog.bind_provider('px', px)
catalog.py = py
self.assertIs(catalog.px, px)
self.assertIs(catalog.get_provider('px'), px)
self.assertIs(catalog.py, py)
self.assertIs(catalog.get_provider('py'), py)
def test_bind_provider_with_invalid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DynamicCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = providers.Provider()
catalog = SomeCatalog()
with self.assertRaises(errors.Error):
catalog.bind_provider('px', px)
with self.assertRaises(errors.Error):
catalog.px = px
with self.assertRaises(errors.Error):
catalog.bind_providers(dict(px=px))
def test_bind_providers(self):
"""Test setting of provider via bind_providers() to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.bind_providers(dict(px=px, py=py))
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_setattr(self):
"""Test setting of providers via attributes to catalog."""
px = providers.Provider()
py = providers.Provider()
self.catalog.px = px
self.catalog.py = py
self.assertIs(self.catalog.px, px)
self.assertIs(self.catalog.get_provider('px'), px)
self.assertIs(self.catalog.py, py)
self.assertIs(self.catalog.get_provider('py'), py)
def test_unbind_provider(self):
"""Test that catalog unbinds provider correct."""
self.catalog.px = providers.Provider()
self.catalog.unbind_provider('px')
self.assertFalse(self.catalog.has_provider('px'))
def test_unbind_via_delattr(self):
"""Test that catalog unbinds provider correct."""
self.catalog.px = providers.Provider()
del self.catalog.px
self.assertFalse(self.catalog.has_provider('px'))
def test_provider_is_bound(self):
"""Test that providers are bound to the catalogs."""
self.assertTrue(self.catalog.is_provider_bound(self.catalog.p1))
self.assertEquals(
self.catalog.get_provider_bind_name(self.catalog.p1), 'p1')
self.assertTrue(self.catalog.is_provider_bound(self.catalog.p2))
self.assertEquals(
self.catalog.get_provider_bind_name(self.catalog.p2), 'p2')
def test_provider_binding_to_different_catalogs(self):
"""Test that provider could be bound to different catalogs."""
p1 = self.catalog.p1
p2 = self.catalog.p2
catalog_a = catalogs.DynamicCatalog(pa1=p1, pa2=p2)
catalog_b = catalogs.DynamicCatalog(pb1=p1, pb2=p2)
self.assertTrue(self.catalog.is_provider_bound(p1))
self.assertTrue(catalog_a.is_provider_bound(p1))
self.assertTrue(catalog_b.is_provider_bound(p1))
self.assertEquals(self.catalog.get_provider_bind_name(p1), 'p1')
self.assertEquals(catalog_a.get_provider_bind_name(p1), 'pa1')
self.assertEquals(catalog_b.get_provider_bind_name(p1), 'pb1')
self.assertTrue(self.catalog.is_provider_bound(p2))
self.assertTrue(catalog_a.is_provider_bound(p2))
self.assertTrue(catalog_b.is_provider_bound(p2))
self.assertEquals(self.catalog.get_provider_bind_name(p2), 'p2')
self.assertEquals(catalog_a.get_provider_bind_name(p2), 'pa2')
self.assertEquals(catalog_b.get_provider_bind_name(p2), 'pb2')
def test_provider_rebinding_to_the_same_catalog(self):
"""Test provider rebinding to the same catalog."""
with self.assertRaises(errors.Error):
self.catalog.p3 = self.catalog.p1
def test_provider_binding_with_the_same_name(self):
"""Test binding of provider with the same name."""
with self.assertRaises(errors.Error):
self.catalog.bind_provider('p1', providers.Provider())
def test_get(self):
"""Test getting of providers using get() method."""
self.assertIs(self.catalog.get_provider('p1'), self.catalog.p1)
self.assertIs(self.catalog.get_provider('p2'), self.catalog.p2)
def test_get_undefined(self):
"""Test getting of undefined providers using get() method."""
with self.assertRaises(errors.UndefinedProviderError):
self.catalog.get_provider('undefined')
with self.assertRaises(errors.UndefinedProviderError):
self.catalog.undefined
def test_has_provider(self):
"""Test checks of providers availability in catalog."""
self.assertTrue(self.catalog.has_provider('p1'))
self.assertTrue(self.catalog.has_provider('p2'))
self.assertFalse(self.catalog.has_provider('undefined'))
def test_filter_all_providers_by_type(self):
"""Test getting of all catalog providers of specific type."""
self.assertTrue(len(self.catalog.filter(providers.Provider)) == 2)
self.assertTrue(len(self.catalog.filter(providers.Value)) == 0)
def test_repr(self):
"""Test catalog representation."""
self.assertIn('TestCatalog', repr(self.catalog))
self.assertIn('p1', repr(self.catalog))
self.assertIn('p2', repr(self.catalog))
class DeclarativeCatalogTests(unittest.TestCase):
"""Declarative catalog tests."""
def test_cls_providers(self):
"""Test `di.DeclarativeCatalog.cls_providers` contents."""
self.assertDictEqual(CatalogA.cls_providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
self.assertDictEqual(CatalogB.cls_providers,
dict(p21=CatalogB.p21,
p22=CatalogB.p22))
def test_inherited_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(CatalogA.inherited_providers, dict())
self.assertDictEqual(CatalogB.inherited_providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
def test_providers(self):
"""Test `di.DeclarativeCatalog.inherited_providers` contents."""
self.assertDictEqual(CatalogA.providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12))
self.assertDictEqual(CatalogB.providers,
dict(p11=CatalogA.p11,
p12=CatalogA.p12,
p21=CatalogB.p21,
p22=CatalogB.p22))
def test_bind_provider(self):
"""Test setting of provider via bind_provider() to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogA.bind_provider('px', px)
CatalogA.bind_provider('py', py)
self.assertIs(CatalogA.px, px)
self.assertIs(CatalogA.get_provider('px'), px)
self.assertIs(CatalogA.py, py)
self.assertIs(CatalogA.get_provider('py'), py)
del CatalogA.px
del CatalogA.py
def test_bind_provider_with_valid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DeclarativeCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = SomeProvider()
py = SomeProvider()
SomeCatalog.bind_provider('px', px)
SomeCatalog.py = py
self.assertIs(SomeCatalog.px, px)
self.assertIs(SomeCatalog.get_provider('px'), px)
self.assertIs(SomeCatalog.py, py)
self.assertIs(SomeCatalog.get_provider('py'), py)
def test_bind_provider_with_invalid_provided_type(self):
"""Test setting of provider with provider type restriction."""
class SomeProvider(providers.Provider):
"""Some provider."""
class SomeCatalog(catalogs.DeclarativeCatalog):
"""Some catalog with provider type restriction."""
provider_type = SomeProvider
px = providers.Provider()
with self.assertRaises(errors.Error):
SomeCatalog.bind_provider('px', px)
with self.assertRaises(errors.Error):
SomeCatalog.px = px
with self.assertRaises(errors.Error):
SomeCatalog.bind_providers(dict(px=px))
def test_bind_providers(self):
"""Test setting of provider via bind_providers() to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogB.bind_providers(dict(px=px, py=py))
self.assertIs(CatalogB.px, px)
self.assertIs(CatalogB.get_provider('px'), px)
self.assertIs(CatalogB.py, py)
self.assertIs(CatalogB.get_provider('py'), py)
del CatalogB.px
del CatalogB.py
def test_setattr(self):
"""Test setting of providers via attributes to catalog."""
px = providers.Provider()
py = providers.Provider()
CatalogB.px = px
CatalogB.py = py
self.assertIs(CatalogB.px, px)
self.assertIs(CatalogB.get_provider('px'), px)
self.assertIs(CatalogB.py, py)
self.assertIs(CatalogB.get_provider('py'), py)
del CatalogB.px
del CatalogB.py
def test_unbind_provider(self):
"""Test that catalog unbinds provider correct."""
CatalogB.px = providers.Provider()
CatalogB.unbind_provider('px')
self.assertFalse(CatalogB.has_provider('px'))
def test_unbind_via_delattr(self):
"""Test that catalog unbinds provider correct."""
CatalogB.px = providers.Provider()
del CatalogB.px
self.assertFalse(CatalogB.has_provider('px'))
def test_provider_is_bound(self):
"""Test that providers are bound to the catalogs."""
self.assertTrue(CatalogA.is_provider_bound(CatalogA.p11))
self.assertEquals(CatalogA.get_provider_bind_name(CatalogA.p11), 'p11')
self.assertTrue(CatalogA.is_provider_bound(CatalogA.p12))
self.assertEquals(CatalogA.get_provider_bind_name(CatalogA.p12), 'p12')
def test_provider_binding_to_different_catalogs(self):
"""Test that provider could be bound to different catalogs."""
p11 = CatalogA.p11
p12 = CatalogA.p12
class CatalogD(catalogs.DeclarativeCatalog):
"""Test catalog."""
pd1 = p11
pd2 = p12
class CatalogE(catalogs.DeclarativeCatalog):
"""Test catalog."""
pe1 = p11
pe2 = p12
self.assertTrue(CatalogA.is_provider_bound(p11))
self.assertTrue(CatalogD.is_provider_bound(p11))
self.assertTrue(CatalogE.is_provider_bound(p11))
self.assertEquals(CatalogA.get_provider_bind_name(p11), 'p11')
self.assertEquals(CatalogD.get_provider_bind_name(p11), 'pd1')
self.assertEquals(CatalogE.get_provider_bind_name(p11), 'pe1')
self.assertTrue(CatalogA.is_provider_bound(p12))
self.assertTrue(CatalogD.is_provider_bound(p12))
self.assertTrue(CatalogE.is_provider_bound(p12))
self.assertEquals(CatalogA.get_provider_bind_name(p12), 'p12')
self.assertEquals(CatalogD.get_provider_bind_name(p12), 'pd2')
self.assertEquals(CatalogE.get_provider_bind_name(p12), 'pe2')
def test_provider_rebinding_to_the_same_catalog(self):
"""Test provider rebinding to the same catalog."""
with self.assertRaises(errors.Error):
class TestCatalog(catalogs.DeclarativeCatalog):
"""Test catalog."""
p1 = providers.Provider()
p2 = p1
def test_provider_rebinding_to_the_same_catalogs_hierarchy(self):
"""Test provider rebinding to the same catalogs hierarchy."""
class TestCatalog1(catalogs.DeclarativeCatalog):
"""Test catalog."""
p1 = providers.Provider()
with self.assertRaises(errors.Error):
class TestCatalog2(TestCatalog1):
"""Test catalog."""
p2 = TestCatalog1.p1
def test_get(self):
"""Test getting of providers using get() method."""
self.assertIs(CatalogB.get('p11'), CatalogB.p11)
self.assertIs(CatalogB.get('p12'), CatalogB.p12)
self.assertIs(CatalogB.get('p22'), CatalogB.p22)
self.assertIs(CatalogB.get('p22'), CatalogB.p22)
self.assertIs(CatalogB.get_provider('p11'), CatalogB.p11)
self.assertIs(CatalogB.get_provider('p12'), CatalogB.p12)
self.assertIs(CatalogB.get_provider('p22'), CatalogB.p22)
self.assertIs(CatalogB.get_provider('p22'), CatalogB.p22)
def test_get_undefined(self):
"""Test getting of undefined providers using get() method."""
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.get('undefined')
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.get_provider('undefined')
with self.assertRaises(errors.UndefinedProviderError):
CatalogB.undefined
def test_has(self):
"""Test checks of providers availability in catalog."""
self.assertTrue(CatalogB.has('p11'))
self.assertTrue(CatalogB.has('p12'))
self.assertTrue(CatalogB.has('p21'))
self.assertTrue(CatalogB.has('p22'))
self.assertFalse(CatalogB.has('undefined'))
self.assertTrue(CatalogB.has_provider('p11'))
self.assertTrue(CatalogB.has_provider('p12'))
self.assertTrue(CatalogB.has_provider('p21'))
self.assertTrue(CatalogB.has_provider('p22'))
self.assertFalse(CatalogB.has_provider('undefined'))
def test_filter_all_providers_by_type(self):
"""Test getting of all catalog providers of specific type."""
self.assertTrue(len(CatalogB.filter(providers.Provider)) == 4)
self.assertTrue(len(CatalogB.filter(providers.Value)) == 0)
def test_repr(self):
"""Test catalog representation."""
self.assertIn('CatalogA', repr(CatalogA))
self.assertIn('p11', repr(CatalogA))
self.assertIn('p12', repr(CatalogA))
self.assertIn('CatalogB', repr(CatalogB))
self.assertIn('p11', repr(CatalogB))
self.assertIn('p12', repr(CatalogB))
self.assertIn('p21', repr(CatalogB))
self.assertIn('p22', repr(CatalogB))
def test_abstract_catalog_backward_compatibility(self):
"""Test that di.AbstractCatalog is available."""
self.assertIs(catalogs.DeclarativeCatalog, catalogs.AbstractCatalog)
class OverrideTests(unittest.TestCase):
"""Catalog overriding and override decorator test cases."""
def tearDown(self):
"""Tear test environment down."""
CatalogA.reset_override()
def test_overriding(self):
"""Test catalog overriding with another catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
self.assertEqual(len(CatalogA.overridden_by), 1)
def test_override_declarative_catalog_with_itself(self):
"""Test catalog overriding of declarative catalog with itself."""
with self.assertRaises(errors.Error):
CatalogA.override(CatalogA)
def test_override_declarative_catalog_with_subclass(self):
"""Test catalog overriding of declarative catalog with subclass."""
with self.assertRaises(errors.Error):
CatalogB.override(CatalogA)
def test_override_dynamic_catalog_with_itself(self):
"""Test catalog overriding of dynamic catalog with itself."""
catalog = catalogs.DynamicCatalog(p11=providers.Value(1),
p12=providers.Value(2))
with self.assertRaises(errors.Error):
catalog.override(catalog)
def test_overriding_with_dynamic_catalog(self):
"""Test catalog overriding with another dynamic catalog."""
CatalogA.override(catalogs.DynamicCatalog(p11=providers.Value(1),
p12=providers.Value(2)))
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
self.assertEqual(len(CatalogA.overridden_by), 1)
def test_is_overridden(self):
"""Test catalog is_overridden property."""
self.assertFalse(CatalogA.is_overridden)
@catalogs.override(CatalogA)
class OverridingCatalog(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
self.assertTrue(CatalogA.is_overridden)
def test_last_overriding(self):
"""Test catalog last_overriding property."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
self.assertIs(CatalogA.last_overriding, OverridingCatalog2)
def test_last_overriding_on_not_overridden(self):
"""Test catalog last_overriding property on not overridden catalog."""
self.assertIsNone(CatalogA.last_overriding)
def test_reset_last_overriding(self):
"""Test resetting last overriding catalog."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(3)
p12 = providers.Value(4)
CatalogA.reset_last_overriding()
self.assertEqual(CatalogA.p11(), 1)
self.assertEqual(CatalogA.p12(), 2)
def test_reset_last_overriding_when_not_overridden(self):
"""Test resetting last overriding catalog when it is not overridden."""
with self.assertRaises(errors.Error):
CatalogA.reset_last_overriding()
def test_reset_override(self):
"""Test resetting all catalog overrides."""
@catalogs.override(CatalogA)
class OverridingCatalog1(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(1)
p12 = providers.Value(2)
@catalogs.override(CatalogA)
class OverridingCatalog2(catalogs.DeclarativeCatalog):
"""Overriding catalog."""
p11 = providers.Value(3)
p12 = providers.Value(4)
CatalogA.reset_override()
self.assertFalse(CatalogA.p11.is_overridden)
self.assertFalse(CatalogA.p12.is_overridden)
class CatalogModuleBackwardCompatibility(unittest.TestCase):
"""Backward compatibility test of catalog module."""
def test_import_catalog(self):
"""Test that module `catalog` is the same as `catalogs`."""
from dependency_injector import catalog
from dependency_injector import catalogs
self.assertIs(catalog, catalogs)
class TestCatalogWithProvidingCallbacks(unittest.TestCase):
"""Catalog with providing callback tests."""
def test_concept(self):
"""Test concept."""
class UsersService(object):
"""Users service, that has dependency on database."""
class AuthService(object):
"""Auth service, that has dependencies on users service."""
def __init__(self, users_service):
"""Initializer."""
self.users_service = users_service
class Services(catalogs.DeclarativeCatalog):
"""Catalog of service providers."""
@providers.Factory
def users():
"""Provide users service.
:rtype: providers.Provider -> UsersService
"""
return UsersService()
@providers.Factory
@injections.inject(users_service=users)
def auth(**kwargs):
"""Provide users service.
:rtype: providers.Provider -> AuthService
"""
return AuthService(**kwargs)
# Retrieving catalog providers:
users_service = Services.users()
auth_service = Services.auth()
# Making some asserts:
self.assertIsInstance(auth_service.users_service, UsersService)
self.assertIsNot(users_service, Services.users())
self.assertIsNot(auth_service, Services.auth())
# Overriding auth service provider and making some asserts:
class ExtendedAuthService(AuthService):
"""Extended version of auth service."""
def __init__(self, users_service, ttl):
"""Initializer."""
self.ttl = ttl
super(ExtendedAuthService, self).__init__(
users_service=users_service)
class OverriddenServices(Services):
"""Catalog of service providers."""
@providers.override(Services.auth)
@providers.Factory
@injections.inject(users_service=Services.users)
@injections.inject(ttl=3600)
def auth(**kwargs):
"""Provide users service.
:rtype: providers.Provider -> AuthService
"""
return ExtendedAuthService(**kwargs)
auth_service = Services.auth()
self.assertIsInstance(auth_service, ExtendedAuthService)