Source code for glorpen.di.container

# -*- coding: utf-8 -*-
'''

.. moduleauthor:: Arkadiusz Dzięgiel <arkadiusz.dziegiel@glorpen.pl>

'''
import inspect
import functools
import importlib
import six

from glorpen.di import exceptions
from glorpen.di.scopes import ScopePrototype, ScopeSingleton, ScopeBase

try:
    from inspect import signature
    signature_empty = inspect.Parameter.empty
except ImportError:
    from funcsigs import signature
    from funcsigs import _empty as signature_empty

[docs]def fluid(f): """Decorator for applying fluid pattern to class methods and to disallow calling when instance is marked as frozen. """ @functools.wraps(f) def wrapper(self, *args, **kwargs): if self._frozen: raise exceptions.ServiceAlreadyCreated(self.name) else: f(self, *args, **kwargs) return self return wrapper
[docs]def normalize_name(o): """Gets object "name" for use with :class:`.Container`. Args: o (object): Object to get name for Returns: str Raises: Exception """ if inspect.isclass(o) or inspect.isfunction(o): return "%s.%s" % (o.__module__, o.__name__) if isinstance(o, str): return o raise Exception("Unknown object: %r" % o)
[docs]class Deffered(object): """Class for marking values for lazy resolving. Values are resolved by :class:`.Container` upon service creation. """ def __init__(self, service=None, method=None, param=None): super(Deffered, self).__init__() self.service = service self.method = method self.param = param
[docs] def resolve(self, getter, param_getter): """Given (service) getter and param_getter, returns resolved value.""" if self.service: svc = getter(self.service) if self.method: return getattr(svc, self.method) else: return svc if self.param: return param_getter(self.param) raise Exception()
[docs]class Kwargs(object): """Simply wraps given kwargs for later use.""" def __init__(self, **kwargs): super(Kwargs, self).__init__() self.kwargs = kwargs
[docs] @classmethod def merge(cls, *args): """Merges iterable arguments, can be `dict` or :class:`.Kwargs` instance. @return: Resulting dict """ ret = {} for arg in args: if arg is None: continue if isinstance(arg, cls): arg = arg.kwargs ret.update(arg) return ret
[docs]class Service(object): """Service definition. When filling arguments for constructor and method calls you can use: - `my_var__svc=MyClass` - will inject service MyClass to `my_var` - `my_var__param="my.param"` - will inject parameter named "my.param" to `my_var` Implementation value for service can be: - class instance - string with import path - callable """ _impl = None _name_or_impl = None _factory = None _scope = ScopeSingleton _load_signature = False _frozen = False def __init__(self, name_or_impl): super(Service, self).__init__() self._kwargs = {} self._sets = {} self._calls = [] self._configurators = [] self._kwargs_modifiers = [] self.name = normalize_name(name_or_impl) self._name_or_impl = name_or_impl def _get_implementation(self): if self._impl: return self._impl if callable(self._name_or_impl): return self._name_or_impl else: return self._lazy_import(self._name_or_impl)() raise exceptions.ContainerException("Bad implementation argument for %r service" % self.name) def _lazy_import(self, path): """Wraps import path in callable returning class object""" module, cls = path.rsplit(".", 1) @functools.wraps(self._lazy_import) def wrapper(*args, **kwargs): return getattr(importlib.import_module(module), cls) return wrapper def _deffer(self, ret=None, svc=None, method=None, param=None): """Wraps value in :class:`.Deffered`. If *ret* argument is given it is returned unchanged.""" if not ret is None: return ret elif svc or param: return Deffered(service=svc, method=method, param=param)
[docs] @fluid def implementation(self, v): """Sets service implementation (callable). Returns: :class:`.Service` """ self._impl = v
[docs] @fluid def factory(self, service=None, method=None, callable=None, kwargs=None, **kwargs_inline): """Sets factory callable. Returns: :class:`.Service`""" self._factory = (self._deffer(svc=service, method=method, ret=callable), self._normalize_kwargs(kwargs_inline, kwargs))
def _normalize_kwargs(self, *args): kwargs = Kwargs.merge(*args) kw = {} for k,v in kwargs.items(): if k.endswith("__svc"): kw[k[:-5]] = self._deffer(svc=v) elif k.endswith("__param"): kw[k[:-7]] = self._deffer(param=v) else: kw[k]=v return kw
[docs] @fluid def kwargs(self, **kwargs): """Adds service constructor arguments. Returns: :class:`.Service`""" self._kwargs.update(self._normalize_kwargs(kwargs))
[docs] @fluid def call(self, method, kwargs=None, **kwargs_inline): """Adds a method call after service creation with given arguments. Returns: :class:`.Service`""" self._calls.append((False, method, self._normalize_kwargs(kwargs_inline, kwargs)))
[docs] @fluid def call_with_signature(self, method, kwargs=None, **kwargs_inline): """Adds a method call after service creation with given arguments. Arguments detected from function signature are added if not already present. Returns: :class:`.Service`""" self._calls.append((True, method, self._normalize_kwargs(kwargs_inline, kwargs)))
[docs] @fluid def set(self, **kwargs): """Will :func:`setattr` given arguments on service. Returns: :class:`.Service`""" self._sets.update(self._normalize_kwargs(kwargs))
[docs] @fluid def configurator(self, service=None, method=None, callable=None, kwargs=None, **kwargs_inline): """Adds service or callable as configurator of this service. Args: service + method, callable: given service method/callable will be called with instance of this service. Returns: :class:`.Service` """ if method or callable: self._configurators.append((self._deffer(svc=service, method=method, ret=callable), self._normalize_kwargs(kwargs_inline, kwargs)))
[docs] @fluid def kwargs_modifier(self, service=None, method=None, callable=None, kwargs=None, **kwargs_inline): """Adds service or callable as constructor arguments modifier of this service. Args: service + method, callable: given service method/callable will be called with kwargs of this service. Returns: :class:`.Service` """ if method or callable: self._kwargs_modifiers.append((self._deffer(svc=service, method=method, ret=callable), self._normalize_kwargs(kwargs_inline, kwargs)))
[docs] @fluid def kwargs_from_signature(self): """Adds arguments found in class signature, based on provided function hints. Returns: :class:`.Service` """ self._load_signature = True
[docs] @fluid def scope(self, scope_cls): """Sets service scope. Returns: :class:`.Service` """ self._scope = scope_cls
[docs]class Alias(object): """Alias for service.""" def __init__(self, target): super(Alias, self).__init__() self.target = normalize_name(target)
[docs]class Container(object): """Implementation of DIC container.""" scopes_cls = [] scopes = [] def __init__(self): super(Container, self).__init__() self.services = {} self.parameters = {} self.self_service_name = normalize_name(self.__class__) self.set_scope_hierarchy(ScopeSingleton, ScopePrototype)
[docs] def set_scope_hierarchy(self, *scopes): """Sets used scopes hierarchy. Arguments should be scopes sorted from widest to narrowest. A service in wider scope cannot request service from narrower one. Default is: [:class:`glorpen.di.scopes.ScopeSingleton`, :class:`glorpen.di.scopes.ScopePrototype`]. Args: classes or instances of :class:`glorpen.di.scopes.ScopeBase` """ my_scopes = [] my_scopes_cls = [] for scope in scopes: if isinstance(scope, ScopeBase): my_scopes.append(scope) my_scopes_cls.append(scope.__class__) else: my_scopes.append(scope()) my_scopes_cls.append(scope) self.scopes = tuple(my_scopes) self.scopes_cls = dict([(o,i) for i,o in enumerate(tuple(my_scopes_cls))])
[docs] def add_service(self, name): """Adds service definition to this container. *name* argument should be a class, import path, or string if :meth:`.Service.implementation` will be used. Returns: :class:`.Service` """ s = Service(name) self.services[s.name] = s return s
[docs] def add_alias(self, service, alias): """Adds an alias for given service""" a = Alias(service) if not a.target in self.services or not isinstance(self.services[a.target], Service): raise exceptions.InvalidAliasTargetException(a.target) self.services[alias] = a return a
[docs] def add_parameter(self, name, value): """Adds a key-value parameter.""" self.parameters[name] = value
[docs] def get(self, svc): """Gets service instance. Raises: UnkownServiceException """ try: return self._get(svc) except exceptions.ContainerException as e: six.reraise(e.__class__, e)
[docs] def get_parameter(self, name): """Gets parameter. Raises: UnkownParameterException """ if name in self.parameters: return self.parameters[name] else: raise exceptions.UnknownParameterException(name)
[docs] def get_definition(self, svc): """Returns definition for given service name.""" name = normalize_name(svc) if not name in self.services: raise exceptions.UnknownServiceException(name) return self.services[name]
def _get_service_definition(self, name): s = self.services.get(name) if hasattr(s, "target"): return self.services.get(s.target) return s def _get(self, svc, requester_chain=None): name = normalize_name(svc) if name == self.self_service_name: return self if not name in self.services: raise exceptions.UnknownServiceException(name) s_def = self._get_service_definition(name) my_scope = s_def._scope if not my_scope in self.scopes_cls: raise exceptions.UnknownScopeException(my_scope, s_def) scope_index = self.scopes_cls[my_scope] if not requester_chain: requester_chain = [] else: requester_scope = requester_chain[-1]._scope if requester_scope and scope_index > self.scopes_cls[requester_scope]: raise exceptions.ScopeWideningException(s_def, requester_chain) def resolver(value): if isinstance(value, Deffered): if s_def in requester_chain: raise exceptions.RecursionException(s_def, requester_chain) return value.resolve(lambda name:self._get(name, requester_chain + [s_def]), self.get_parameter) else: return value def service_creator(): return self._create(s_def, resolver) return self.scopes[scope_index].get(service_creator, name) def _update_kwargs_from_signature(self, function, kwargs): try: sig = signature(function) except ValueError: return for name, param in tuple(sig.parameters.items()): if name == "self": continue if param.annotation is signature_empty: continue n = normalize_name(param.annotation) if n in self.services: kwargs.setdefault(name, Deffered(service=n)) def _create(self, s_def, resolver): s_def._frozen = True def resolve_kwargs(kwargs): return dict([(k, resolver(v)) for k,v in kwargs.items()]) if s_def._factory: cls = resolver(s_def._factory[0]) kwargs = s_def._factory[1] else: cls = s_def._get_implementation() kwargs = {} kwargs.update(s_def._kwargs) self._update_kwargs_from_signature(cls.__init__, kwargs) kwargs = resolve_kwargs(kwargs) for conf, params in s_def._kwargs_modifiers: resolver(conf)(kwargs, **resolve_kwargs(params)) try: instance = cls(**kwargs) except Exception as e: six.raise_from(exceptions.InjectionException(s_def.name, cls), e) for conf, params in s_def._configurators: resolver(conf)(instance, **resolve_kwargs(params)) for k,v in resolve_kwargs(s_def._sets).items(): setattr(instance, k, v) for use_sig, call_method, call_kwargs in s_def._calls: callable = getattr(instance, call_method) kwargs = dict(call_kwargs) if use_sig: self._update_kwargs_from_signature(callable, kwargs) try: callable(**resolve_kwargs(kwargs)) except Exception as e: six.raise_from(exceptions.InjectionException(s_def.name, cls, call_method), e) return instance