import abc
import asyncio
import traceback
import types
from aiolocals.local import _get_ident as get_ident
def dependency_id_to_string(dep_id):
"""
Convert the dependency ID to string
:param dep_id: Dependency ID, the first argument of di.Dep
:rtype: str
"""
if isinstance(dep_id, str):
return dep_id
elif hasattr(dep_id, '__name__'):
return dep_id.__name__
else:
raise RuntimeError("Dependency ID can be either string or classobject")
[docs]def has_deps(cls):
"""
Decorate with me the classes that have dependencies.
Usage example:
..code::
@di.has_deps
class Depends:
dep = di.Dep(...)
"""
deps = {}
for attr_name in dir(cls):
if not attr_name.startswith('__') and not attr_name.endswith('__'):
value = getattr(cls, attr_name)
if isinstance(value, Dep):
deps[attr_name] = value
setattr(cls, '__dependencies__', deps)
__old_init__ = getattr(cls, '__init__')
def __init__(self, *args, **kwargs):
"""
Create new instance and perform the actual injection.
:param _cls: The class which was decorated with @has_deps
:rtype: _cls
"""
nonlocal __old_init__
_container = get_container_for_ident()
_container.inject(self)
__old_init__(self, *args, **kwargs)
setattr(cls, '__init__', __init__)
return cls
[docs]class Dep:
"""
A dependency to be injected.
Usage example:
.. code::
@di.has_deps
class Depends:
mysql = di.Dep(Mysql)
def fetch_data(self):
self.mysql.run_query(*args) ...
"""
def __init__(self, _type, name=None):
self.type = _type
self.name = name or _type.__name__
def __get__(self, instance, owner):
if instance is None:
return self
raise RuntimeError(
"Dependency named {0} depended on by {1!r} is not configured properly, ".format(
self.name, owner
) + "probably @di.has_deps decorator is missing on dependent class"
)
def __repr__(self):
return "<Dependency: {!s}>".format(self.name)
[docs]class Container:
"""
A container for dependencies.
"""
def __init__(self):
# Map of dep key to _Contained
self.__provides = {}
self.preserved = {}
class _Contained:
def __init__(self, name, instance):
self.name = name
self.instance = instance
self.creation_point = traceback.extract_stack()
def __repr__(self):
return "<{name} instance: {instance!r}>".format(name=self.name, instance=self.instance)
def to_msg(self):
return "{!r} installed here:\n{}".format(
self,
"".join(traceback.format_list(self.creation_point[:-3])))
def inject(self, instance):
deps = getattr(instance, '__dependencies__')
if deps:
for attr, dep in deps.items():
contained = self.__provides[dep.name]
if contained.instance is not None and not isinstance(contained.instance, dep.type):
raise TypeError("Wrong provider for {!r}".format(contained.instance))
setattr(instance, attr, contained.instance)
return instance
def clone(self):
container = Container()
for (k, v) in self.items():
container[k] = v
return container
def replace(self, name_or_class, instance, component=None):
"""
Forces the replacement of name_or_class in container.
:param name_or_class: The class or name (string) for the dependency key
:param instance: the new instance to use to provide name_or_class
:param component: deprecated param name. Use instance instead.
"""
# TODO: remove component param once it is unused
if component is not None and instance is not None:
raise AttributeError("Should not specify both component and instance parameters. Use instance.")
if component is not None:
instance = component
string_dep_id = dependency_id_to_string(name_or_class)
if string_dep_id not in self.__provides:
raise RuntimeError("An attempt to replace non-existent dependency provider")
elif not isinstance(instance, self.__provides[string_dep_id].instance.__class__):
raise RuntimeError("An attempt to replace dependency provider with instance having different class")
self.__provides[string_dep_id] = self._Contained(string_dep_id, instance)
def __delitem__(self, key):
string_dep_id = dependency_id_to_string(key)
del self.__provides[string_dep_id]
def __setitem__(self, name_or_class, component):
string_dep_id = dependency_id_to_string(name_or_class)
contained = self._Contained(string_dep_id, component)
if string_dep_id in self.__provides:
# TODO: save the traceback for every install so we can give both here.
raise RuntimeError(self._format_two_contained(self.__provides[string_dep_id], contained))
self.__provides[string_dep_id] = contained
def __getitem__(self, name_or_class):
string_dep_id = dependency_id_to_string(name_or_class)
return self.__provides.get(string_dep_id).instance
def items(self):
"""
Returns container members alongside with keys
:rtype: collections.Iterable
"""
return [(x[0], x[1].instance) for x in self.__provides.items()]
[docs] def values(self):
"""
Returns container members as dictionary values.
"""
return [x.instance for x in self.__provides.values()]
def keys(self):
"""
Return keys of provider dictionary
"""
return self.__provides.keys()
[docs] def reset(self):
"""
Suitable when recreating component environment in test cases setUp/tearDown.
"""
self.__provides.clear()
@staticmethod
def _format_two_contained(first, second):
return "Trying to install two dependency providers with the same name.\n" \
"First {first}\nSecond {second}".format(first=first.to_msg(), second=second.to_msg())
# A mapping of context-local containers. This is global process state.
contextual_containers = {}
def get_container_for_ident():
"""
Gets the container for the current ident. Defaults
:rtype: micro.di.Container
"""
global contextual_containers
return contextual_containers.get(get_ident(), container)
default_container = Container()
# This is default container and the one which plays for non-local contexts.
container = Container()
contextual_containers[0] = container
[docs]class Component(type):
"""
***
Deprecated: New code doesn't need to use this class. Instead, simply declaring a
start() method or coroutine is enough to be started.
***
A metaclass for components which can be injected as dependencies themselves.
Components must declare a start() method or coroutine.
"""
def __init__(cls, name, bases, dct):
if 'start' not in dct or not isinstance(dct['start'], types.FunctionType):
raise ValueError("Component class {!r} must declare start() method.".format(cls))
super(Component, cls).__init__(name, bases, dct)
[docs]def provide(name_or_class, instance):
"""
Provide dependency for name or class to the task local scope.
"""
_container = get_container_for_ident()
_container[dependency_id_to_string(name_or_class)] = instance
@asyncio.coroutine
def init_components():
"""
Initialize the registered components.
Any object that has been di.Provide'd that has a start() method will have its start method called.
If a start method is a coroutine or returns a coroutine, it will be called.
Otherwise, the return value of a start method is ignored.
Note: because this is itself a coroutine you must use `yield from di.init_components()` or similar to actually execute it.
"""
for instance in container.values():
if hasattr(instance, 'start') and callable(instance.start):
c = instance.start()
if asyncio.iscoroutine(c):
yield from c
[docs]class DIContext(metaclass=abc.ABCMeta):
def __init__(self, replace_dict, loop=None):
"""
Initialize DIContext. Should be used like
.. code::
from asynctest import mock
from dblib.redis import Redis
from dblib.MySQL import MySQL
with DIContext({'MySQL': mock.Mock(MySQL), Redis: mock.Mock(Redis)}):
.. some code using different providers for various dependencies ..
DIContext itself does not provide any real contextual state modification, instead it defers the implementation
to inheritors.
:param replace_dict: The dict to substitute while in context
:type replace_dict: dict
:param loop: Asyncio event loop or None
This provider context is not task-aware, so you won`t use it for any other
purpose than switching providers for test methods.
"""
if not replace_dict:
raise RuntimeError("Invalid provider subsitution: no dependency providers to substitute")
self.replace_dict = {}
for (item, provider) in replace_dict.items():
self.replace_dict[dependency_id_to_string(item)] = provider
self.preserved = None
if not loop:
loop = asyncio.get_event_loop()
self.loop = loop
self.local_container = None
@abc.abstractmethod
[docs] def __enter__(self, _container=container):
"""
Enter the provider context.
:param _container: The container to substitute (defaults to global one)
:type _container: micro.di.Container
:rtype: micro.di.Container
"""
raise NotImplemented() # pragma: no cover
@abc.abstractmethod
[docs] def __exit__(self, exc_type, exc_val, exc_tb):
"""
Exit the provider context and cleanup the substitutions. Accepts the common __exit__ arguments
"""
raise NotImplemented() # pragma: no cover
[docs]class ProviderContext(DIContext):
"""
This replaces all current provider with new ones. Useful for applications
which need different sets of providers during app lifecycle (for instance, on application finish
you can switch some transport component class that incapsulates some remote resource communication with the class
which actually cleans the resource connection pool). Most suitable for tests, since there only one test instance
running at a time.
"""
def __enter__(self, _container=container):
self.preserved = {}
if _container is not container:
_container = self.local_container = _container.clone()
else:
# Replace on global container
self.local_container = _container
container_keys = list(_container.keys())
for (item, value) in self.replace_dict.items():
if item in container_keys:
self.preserved[item] = _container[item]
else:
raise RuntimeError("An attempt to substitute and item which is not currently provided, key=%s" % item)
_container.reset()
for (item, value) in self.preserved.items():
replace = self.replace_dict.get(item)
if not isinstance(replace, value.__class__):
raise RuntimeError(
"An attempt to subsitute dependency provider with wrong typed component instance,"
" needed_type={!r}, "
" actual_type={!r}".format(value, replace)
)
_container[item] = replace
return _container
def __exit__(self, exc_type, exc_val, exc_tb):
self.local_container.reset()
for (item, value) in self.preserved.items():
self.local_container[item] = self.preserved[item]
[docs]class TaskContainerContext(ProviderContext):
"""
This context substitutes providers for task-local context. Only the task where context entering is happened
will receive the replace_dict providers; other tasks will have different context
"""
[docs] def __enter__(self, _container=container):
"""
Enter the task local context, All the new tasks will have the replace_dict installed for provider,
for newly created instances in this task will have dependency providers subsituted
:param _container: The container to subsitute (defaults to global one)
:type _container: micro.di.Container
:rtype: micro.di.Container
"""
_container = _container.clone()
self.ident = get_ident()
""":type ident: int"""
replaced_container = super().__enter__(_container)
contextual_containers[self.ident] = replaced_container
return replaced_container
def __exit__(self, exc_type, exc_val, exc_tb):
if self.ident in contextual_containers:
contextual_containers.__delitem__(self.ident)
ident = get_ident()
super().__exit__(exc_type, exc_val, exc_tb)
# Check that exit method is not called in different task local context
if ident != self.ident:
raise RuntimeError("Invalid context subtitution - leaving context in different task!")
__all__ = [
"has_deps", "Dep", "init_components", "provide",
"Container", "container", "get_container_for_ident", "DIContext", "ProviderContext", "TaskContainerContext"
]