Source code for sunnydi.ioc

"""
Inversion of Control
====================

Framework for configuring and composing object graphs injecting
their associated dependencies.

Using inversion-of-control rather than manually building object graphs can
reduce an application's maintenance burden.

For the philosophical reasoning behind such an architecture, see Martin
Fowler's [article](http://martinfowler.com/articles/injection.html).

Getting Started
---------------
In order to create an injector, we must first create and configure a
`sunnydi.ioc.Module`. A module defines how instances will
be built and provided to other instances in the object graph.

For our example, we will create a module for our HelloService.

    #!python
    class HelloService(object):
        def hello(self):
            return 'hello'

Now, we create a custom configuration module that extends
`sunnydi.ioc.Module`. In the most simple configuration,
we just bind a contract name to our HelloService class type:

    #!python
    class HelloModule(Module):
        def configure(self):
            self.bind('hello_service')
                .to(HelloService)

We can then create an injector and resolve our HelloService like this:

    #!python
    hello_module = HelloModule()
    injector = hello_module.create_injector()
    hello_service = injector.get('hello_service')

    >>> hello_service.hello()
    'hello'

Advanced Configuration
----------------
More often than not, classes will have dependencies on other classes,
and those classes will have additional dependencies. This results in
potentially large object graphs that becomes very difficult to manage
manually. On top of that, we probably only need to create some classes
once for the lifetime of the application.

The below configuration illustrates how to accomplish this with the
IoC configuration Module:

    #!python
    class GoodbyeService(object):

        # param name matches our binding contract name
        def __init__(self, hello_service):
            self._hello_service = hello_service

        def goodbye(self):
            return '%s, goodbye' % self._hello_service.hello()

    class HelloModule(Module):

        def configure(self):

            # we only ever need one instance of this service
            self.bind('hello_service')
                .to(HelloService)
                .as_singleton()

            # we only ever need one instance of this service
            self.bind('goodbye_service')
                .to(GoodbyeService)
                .as_singleton()

    ...

    hello_module = HelloModule()
    injector = hello_module.create_injector()

    # resolving the service multiple times
    # returns the same instance due to `as_singleton()`
    goodbye_service = injector.get('goodbye_service')
    goodbye_service2 = injector.get('goodbye_service')

    >>> assert goodbye_service == goodbye_service2
    True

    >>> goodbye_service.goodbye()
    'hello, goodbye'

Occasionally, manual configuration of a class is necessary in
whole or in part. In these cases, the module can configure a
factory method to provide the instance, or provide an instance as-is.

    #!python
    class HelloModule(Module):

        def configure(self):

            # new up an instance on our own
            # this instance is de facto singleton
            hello_service = HelloService()
            # additional configuration
            ...
            self.bind('hello_service')
                .to_instance(hello_service)

            # this service uses a factory to create the instance
            # factory can be static, instance, or global function
            # factory can also be marked as singleton
            self.bind('goodbye_service')
                .to_factory(self._create_goodbye_service)
                .as_singleton()

        @staticmethod
        def _create_goodbye_service(hello_service):
            goodbye_service = GoodbyeService(hello_service)
            # additional configuration
            ...
            return goodbye_service

Resolving Instances
-------------------
Class instances can be resolved directly from the injector via their
contract name(s) or class type(s). Multiple contracts may be resolved
by adding additional parameters to the `get()` call.

    #!python

    # get one
    goodbye_service = injector.get('goodbye_service')

    # get many
    (hello_service, goodbye_service) =
    injector.get('hello_service', 'goodbye_service')

    # get can also take a class type
    goodbye_service = injector.get(GoodbyeService)


For CLI applications, resolving the main application class should be the only
call to `get()` necessary (the remaining object graph should be populated
via the injector).

For non-CLI or other applications in which object lifecycle isn't fully
controlled, the `sunnydi.ioc.inject` decorator may be used
on a class's `__init__()` method (`MyClass` does _not_ need to be configured
in the module).

The `sunnydi.ioc.inject` decorator is not necessary for classes
resolved via the injector (only for classes outside the injection context).

    #!python
    class MyClass(object):
        @inject
        def __init__(self, hello_service, goodbye_service):
            self._hello_service = hello_service
            self._goodbye_service = goodbye_service

    # same as calling
    my_class = injector.get(MyClass)

Global Injector
---------------
In some cases, there may be a need to import and use the
`sunnydi.ioc.Injector` from the global context.

    #!python
    from sunnydi.ioc import get_injector

The `sunnydi.ioc.Injector` can also be resolved from the
`sunnydi.ioc.inject` decorator:

    #!python
    @inject
    def __init__(self, injector):
        pass

    # same as calling
    injector = injector.get('injector')

In order to use the global `sunnydi.ioc.get_injector` or the
`sunnydi.ioc.inject` decorator,
we must register the configuration module:

    #!python
    hello_module = HelloModule()
    injector = hello_module.create_injector()
    hello_module.register(injector)

or (if we don't need the `sunnydi.ioc.Injector`
instance right away):

    #!python
    hello_module = HelloModule()
    hello_module.register()

"""

import abc
import uuid
import inspect
import threading
from functools import wraps


__all__ = [
    'DependencyResolutionException',
    'ComponentNotRegisteredException',
    'ScopeDisposedException',
    'Module',
    'Injector',
    'InjectionScope',
    'inject',
    'get',
    'scope',
    'get_injector'
]

_modules = dict()
_injectors = dict()

DEFAULT_INJECTOR = 'Default'


[docs]class ScopeDisposedException(Exception): """ Raised when an `sunnydi.ioc.InjectionScope` has been disposed, but a client has attempted to resolve a component from it. """ pass
[docs]class DependencyResolutionException(Exception): """ Raised when an item could not be resolved from an `sunnydi.ioc.Injector`. """ pass
[docs]class ComponentNotRegisteredException(DependencyResolutionException): """ Raised when a component binding was not registered with an `sunnydi.ioc.Injector`. """ pass
class Binding(object): contract = None class_type = None get_instance = None on_init = None on_cleanup = None injector = None _is_instance_scope = False _is_eager = False _factory_function = None def __init__(self, contract): self.contract = contract def __str__(self): return '%s(contract: "%s", type: "%s")' % ( self.__class__.__name__, self.contract, self.class_type ) class ScopedBinding(Binding): def __init__(self, binding, injector): assert isinstance(binding, Binding) assert isinstance(injector, Injector) super(ScopedBinding, self).__init__(binding.contract) self._lock = threading.Lock() self._instance = None self._is_instance_scope = True self._is_eager = binding._is_eager self.injector = injector self.class_type = binding.class_type self.get_instance = self._get_instance self.on_init = binding.on_init self.on_cleanup = binding.on_cleanup def _get_instance(self): with self._lock: if self._instance is None: injector = self.injector self._instance = injector.get(self.class_type) if self.on_init is not None: self.on_init(self._instance) return self._instance class ScopedFactoryBinding(ScopedBinding): def __init__(self, binding, injector): super(ScopedFactoryBinding, self).__init__(binding, injector) self._factory_function = binding._factory_function self._is_eager = binding._is_eager self.get_instance = self._get_instance self.on_init = binding.on_init self.on_cleanup = binding.on_cleanup def _get_instance(self): with self._lock: if self._instance is None: kwargs = self._resolve_args() self._instance = self._factory_function(**kwargs) if self.on_init is not None: self.on_init(self._instance) return self._instance def _resolve_args(self): injector = self.injector if injector is None: return {} try: arg_spec = inspect.getargspec(self._factory_function) resolved_args = dict() for i, arg in enumerate(arg_spec.args): if arg not in ('self', 'cls'): resolved_args[arg] = injector.get(arg) return resolved_args except TypeError: return {} class ScopedBindingBuilder(object): _binding = None _instance = None def __init__(self, binding): self._binding = binding def _get_instance(self): if self._instance is None: injector = self._binding.injector self._instance = injector.get(self._binding.class_type) return self._instance def as_singleton(self): self._binding.get_instance = self._get_instance return ScopedLifecycleBindingBuilder(self._binding) def instance_per_scope(self): self._binding._is_instance_scope = True self._binding.get_instance = self._get_instance return ScopedLifecycleBindingBuilder(self._binding) class FactoryBindingBuilder(object): _binding = None _instance = None def __init__(self, binding, factory_function): self._binding = binding self._factory_function = factory_function self._is_singleton = False self._binding._factory_function = factory_function self._binding.get_instance = self._get_instance def _get_instance(self): if not self._is_singleton: kwargs = self._resolve_args() return self._factory_function(**kwargs) if self._instance is None: kwargs = self._resolve_args() self._instance = self._factory_function(**kwargs) return self._instance def _resolve_args(self): injector = self._binding.injector if injector is None: return {} try: arg_spec = inspect.getargspec(self._factory_function) resolved_args = dict() for i, arg in enumerate(arg_spec.args): if arg not in ('self', 'cls'): resolved_args[arg] = injector.get(arg) return resolved_args except TypeError: return {} def as_singleton(self): self._is_singleton = True return ScopedLifecycleBindingBuilder(self._binding) def instance_per_scope(self): self._is_singleton = True self._binding._is_instance_scope = True return ScopedLifecycleBindingBuilder(self._binding) class ScopedLifecycleBindingBuilder(object): def __init__(self, binding): self._binding = binding def eager(self): self._binding._is_eager = True return self def on_init(self, init_action): self._binding.on_init = init_action return self def on_cleanup(self, cleanup_action): self._binding.on_cleanup = cleanup_action return self class LinkedBindingBuilder(object): _binding = None def __init__(self, binding): self._binding = binding def _get_instance(self): injector = self._binding.injector return injector.get(self._binding.class_type) def to(self, class_type): self._binding.class_type = class_type self._binding.get_instance = self._get_instance return ScopedBindingBuilder(self._binding) def to_instance(self, obj): self._binding.class_type = type(obj) self._binding.get_instance = lambda: obj def to_factory(self, factory_function): return FactoryBindingBuilder(self._binding, factory_function)
[docs]class Module(object): """ Configuration module for defining dependency-injection bindings. """ __metaclass__ = abc.ABCMeta _name = DEFAULT_INJECTOR def __init__(self, name=None): self._bindings = dict() self._children = dict() if name is not None: self._name = name @property def name(self): """ The module name (or `DEFAULT_INJECTOR`). """ return self._name @abc.abstractmethod
[docs] def configure(self): """ Configure the IoC module, creating any necessary injection bindings. """ raise NotImplementedError()
[docs] def add_module(self, module): """ Add a configuration sub-module to this module. Parameters ---------- * module (`sunnydi.ioc.Module`): A sub-module to add. """ assert isinstance(module, Module) if module.name in self._children: self._children[module.name].add_module(module) else: self._children[module.name] = module
[docs] def bind(self, contract): """ Create a new binding with the specified contract name. Parameters ---------- * contract (basestring): The contract name to bind to. Returns ------- A binding builder. """ binding = Binding(contract) self._bindings[contract] = binding return LinkedBindingBuilder(binding)
[docs] def register(self, injector=None): """ Register the specified `sunnydi.ioc.Injector` with the global scope. If `injector` parameter is not specified, create a new `sunnydi.ioc.Injector` and register it. Parameters ---------- * injector (`sunnydi.ioc.Injector`): The injector to register or create and register a new `sunnydi.ioc.Injector` if None. """ if injector is None: injector = self.create_injector() _modules[self._name] = self _injectors[self._name] = injector
[docs] def create_injector(self): """ Create the dependency `sunnydi.ioc.Injector` using the configured bindings. Returns ------- A configured `sunnydi.ioc.Injector`. """ bindings = self._configure_bindings() injector = Injector(bindings) injector._initialize() return injector
def _configure_bindings(self): self.configure() for n in self._children: module = self._children[n] bindings = module._configure_bindings() self._bindings.update(bindings) return self._bindings
class _ExportsModule(Module): NAME = "DecoratedExports" def __init__(self): super(_ExportsModule, self).__init__(self.NAME) self._injector = None def configure(self): pass def register_once(self): if self.NAME in _injectors: return self._injector = self.create_injector() self.register(self._injector) def update(self): self._injector._update_bindings(self._bindings) class Export(object): """ Decorate a class to register an IoC binding, and inject it into another class. Example: #!python @Export('event_handler') class MyEventHandler(object): pass ... class InjectingClass(object): @inject def __init__(self, event_handler): pass """ _EXPORTS_MODULE = _ExportsModule() def __init__(self, contract, singleton=False): """ Decorate a class to register the class with the IoC container. """ self._contract = contract self._singleton = singleton def __call__(self, f): # create an IoC binding for the decorated class builder = self._EXPORTS_MODULE.bind(self._contract) if self._singleton: builder.to(f).as_singleton() else: builder.to(f) # register the exports module if it isn't already self._EXPORTS_MODULE.register_once() # update our injector with the module changes self._EXPORTS_MODULE.update() # just return the class - we aren't wrapping any functionality return f
[docs]class Injector(object): """ Dependency injector used for resolving dependencies. This class is typically not created explicitly, rather it is created by configuring a `sunnydi.ioc.Module`. """ def __init__(self, bindings=None): self._scope_lock = threading.Lock() self._child_scopes = dict() if bindings is None: self._bindings = dict() else: self._bindings = dict(bindings) # add special binding for self b = Binding('injector') LinkedBindingBuilder(b).to_instance(self) self._bindings['injector'] = b
[docs] def get(self, *args, **kwargs): """ Resolve an instance or instances of the specified contracts. Parameters ---------- * contract (basestring): One or more contract names to resolve. * class_type (type): One or more classes to resolve with constructor parameters injected. * class_args (tuple): (Optional) Collection of arguments to pass to a resolving class's positional arguments (`*args`) instead of resolving parameters via the injector. * class_kwargs (dict): (Optional) Collection of key-word arguments to pass to a resolving class's keyword arguments (`*kwargs`) instead of resolving parameters via the injector. Returns ------- The resolved object instance or tuple of instances if multiple parameters specified. Raises ------ * `sunnydi.ioc.DependencyResolutionException`: If no contracts or types are specified or if `None` type is specified. * `sunnydi.ioc.ComponentNotRegisteredException`: If a binding for the specified contract could not be found. """ # need something to resolve! if len(args) == 0: raise DependencyResolutionException( "No contract(s) specified to resolve" ) resolved = [] for arg in args: # can't resolve null values if arg is None: raise DependencyResolutionException( "Contract must not be None" ) # if the passed arg is a class type, # resolve the class and its constructor arguments if inspect.isclass(arg): class_args = kwargs.get('class_args', tuple()) class_kwargs = kwargs.get('class_kwargs', {}) obj = self._resolve_type( arg, class_args, class_kwargs ) resolved.append(obj) continue # check our binding collection for a # binding that matches the arg contract if arg in self._bindings: binding = self._bindings[arg] resolved.append(binding.get_instance()) else: message = "Could not find binding for contract: '%s'" % arg raise ComponentNotRegisteredException(message) # for single results return just the resolved item # otherwise return a tuple of resolved items if len(resolved) == 1: return resolved[0] return tuple(resolved)
[docs] def is_scope(self, scope_id): """ Whether or not a child `sunnydi.ioc.InjectionScope` with the specified scope id exists for this injector. Will only return `True` for scopes created directly from this injector (not child scopes). Parameters ---------- * scope_id (basestring): The unique scope identifier. Returns ------- `True` if a scope with the specified id exists, `False` if no scope exists. """ with self._scope_lock: return scope_id in self._child_scopes
[docs] def scope(self, scope_id=None): """ Create a new child `sunnydi.ioc.InjectionScope` with the specified scope id or return a previously created child scope. It's recommended to use this method as a context manager in order to properly dispose of the scope when it's finished being used. #!python with injector.scope('my-scope-id') as my_scope: obj = my_scope.get('my_contract_name') If not being used as a context manager, it is mandatory to manually dispose of the scope via the `sunnydi.ioc.InjectionScope.dispose()` method when finished using the scope. Failure to do so will result in memory leaks within the application. #!python my_scope = injector.scope('my-scope-id') obj = my_scope.get('my_contract_name') my_scope.dispose() Parameters ---------- * scope_id (basestring): (Optional) The unique scope identifier. If no scope id is specified, a random `uuid.UUID` is used to create a new scope id. Returns ------- An `sunnydi.ioc.InjectionScope` """ with self._scope_lock: # if no scope id was provided, use a random value if scope_id is None: scope_id = str(uuid.uuid4()) # if the scope id matches an existing scope, # return it - otherwise create a new scope if scope_id in self._child_scopes: return self._child_scopes[scope_id] else: scope = InjectionScope(scope_id, self._bindings, self) self._child_scopes[scope_id] = scope return scope
def _initialize(self): eager_bindings = set() # ensure the injector is available in # the bindings for runtime resolution # also, save our eager bindings for # later initialization for contract in self._bindings: binding = self._bindings[contract] binding.injector = self if binding._is_eager: eager_bindings.add(binding) # now initialize the eager bindings self._initialize_eager(eager_bindings) @staticmethod def _initialize_eager(bindings): for binding in bindings: binding.get_instance() def _update_bindings(self, bindings): self._bindings.update(bindings) def _resolve_type(self, class_type, class_args, class_kwargs): try: arg_spec = inspect.getargspec(class_type.__init__) kwargs = dict() for i, arg in enumerate(arg_spec.args): # skip self/cls if arg in ('self', 'cls'): continue try: # since 'self' or 'cls' is first in the arg spec, # but class_args shouldn't have 'self', we need # to adjust the index if i < len(class_args) + 1: kwargs[arg] = class_args[i - 1] elif arg in class_kwargs: kwargs[arg] = class_kwargs[arg] else: kwargs[arg] = self.get(arg) except BaseException as e: message = "Could not resolve contract: %s, %s" % (arg, e) raise DependencyResolutionException(message) return class_type(**kwargs) except TypeError: try: return class_type() except BaseException as e: message = "Could not resolve class: %s, %s" % (class_type, e) raise DependencyResolutionException(message) def _dispose_child_scope(self, scope): with self._scope_lock: del self._child_scopes[scope._scope_id]
[docs]class InjectionScope(Injector): """ Dependency injector used for resolving dependencies within a limited scope. This class is typically not created explicitly, rather it is created from a parent `sunnydi.ioc.Injector` by calling `scope()`. """ def __init__(self, scope_id, bindings, parent_scope): self._scope_id = scope_id self._parent_scope = parent_scope self._disposed = False # create scoped bindings for # anything marked as 'instance' scope scoped = dict() eager_bindings = set() for key in bindings: binding = bindings[key] if binding._is_instance_scope: # create a new scoped binding if binding._factory_function is not None: scoped[key] = ScopedFactoryBinding(binding, self) else: scoped[key] = ScopedBinding(binding, self) # add scoped binding to eager set for # later initialization if binding._is_eager: eager_bindings.add(scoped[key]) else: # binding not scoped, just use whatever # the parent has defined scoped[key] = binding super(InjectionScope, self).__init__(scoped) self._initialize_eager(eager_bindings) def __str__(self): return "%s('%s')" % ( self.__class__.__name__, self._scope_id ) def __enter__(self): return self def __exit__(self, exception_type, exception_value, traceback): self.dispose() @property def scope_id(self): return self._scope_id
[docs] def get(self, *args, **kwargs): """ Resolve an instance or instances of the specified contracts within the specified scope. Parameters ---------- * contract (basestring): One or more contract names to resolve. * class_type (type): One or more classes to resolve with constructor parameters injected. * class_args (tuple): (Optional) Collection of arguments to pass to a resolving class's positional arguments (`*args`) instead of resolving parameters via the injector. * class_kwargs (dict): (Optional) Collection of key-word arguments to pass to a resolving class's keyword arguments (`*kwargs`) instead of resolving parameters via the injector. Returns ------- The resolved object instance or tuple of instances if multiple parameters specified. Raises ------ * `sunnydi.ioc.DependencyResolutionException`: If no contracts or types are specified or if `None` type is specified. * `sunnydi.ioc.ComponentNotRegisteredException`: If a binding for the specified contract could not be found. * `sunnydi.ioc.ScopeDisposedException`: If the scope has been disposed. """ # ensure this scope hasn't been disposed with self._scope_lock: if self._disposed: raise ScopeDisposedException( "The scope with id: '%s' has " "already been disposed" % self._scope_id ) # otherwise, just resolve as normal return super(InjectionScope, self).get(*args, **kwargs)
[docs] def dispose(self): with self._scope_lock: # nothing to do if we've already disposed if self._disposed: return # call any cleanup actions for our bindings for key in self._bindings: self._dispose_binding(self._bindings[key]) # clear our references self._bindings.clear() self._child_scopes.clear() self._parent_scope._dispose_child_scope(self) self._parent_scope = None self._disposed = True
@staticmethod def _dispose_binding(binding): if binding.on_cleanup is None: return instance = binding.get_instance() binding.on_cleanup(instance)
[docs]def inject(f): """ Inject the dependencies for the decorated function. Each of the function's parameter names should match a configured binding name. If a parameter is included directly, injection is skipped for that parameter. """ @wraps(f) def injector_wrapper(*args, **kwargs): # TODO: decorator param, named injector? injector = get_injector() arg_spec = inspect.getargspec(f) name = f.func_name injected_kwargs = dict() # if the decorator is attached to a property, # inject the service with the name of that property class_type = type(args[0]) p = getattr(class_type, name) is_property = isinstance(p, property) if is_property: return injector.get(name) # for each method argument, attempt to # satisfy injectable dependencies by: # 1). the passed in method argument (if it exists) # 2). the keyword argument (if it exists) # 3). finally, attempt to resolve the argument from the injector for i, arg in enumerate(arg_spec.args): if i < len(args): injected_kwargs[arg] = args[i] elif arg in kwargs: injected_kwargs[arg] = kwargs[arg] else: injected_kwargs[arg] = injector.get(arg) return f(**injected_kwargs) return injector_wrapper
[docs]def get(*args, **kwargs): """ Resolve an instance or instances of the specified contracts. Parameters ---------- * contract (basestring): One or more contract names to resolve. * class_type (type): One or more classes to resolve with constructor parameters injected. * class_args (tuple): (Optional) Collection of arguments to pass to a resolving class's positional arguments (`*args`) instead of resolving parameters via the injector. * class_kwargs (dict): (Optional) Collection of key-word arguments to pass to a resolving class's keyword arguments (`*kwargs`) instead of resolving parameters via the injector. Returns ------- The resolved object instance or tuple of instances if multiple parameters specified. Raises: * `sunnydi.ioc.DependencyResolutionException`: If no contracts or types are specified or if `None` type is specified. * `sunnydi.ioc.ComponentNotRegisteredException`: If a binding for the specified contract could not be found. """ injector = get_injector() return injector.get(*args, **kwargs)
[docs]def scope(scope_id=None): """ Create a new child `sunnydi.ioc.InjectionScope` from the default injector, with the specified scope id or return a previously created child scope. It's recommended to use this method as a context manager in order to properly dispose of the scope when it's finished being used. #!python import ioc with ioc.scope('my-scope-id') as my_scope: obj = my_scope.get('my_contract_name') If not being used as a context manager, it is mandatory to manually dispose of the scope via the `sunnydi.ioc.InjectionScope.dispose()` method when finished using the scope. Failure to do so will result in memory leaks within the application. #!python my_scope = ioc.scope('my-scope-id') obj = my_scope.get('my_contract_name') my_scope.dispose() Parameters ---------- * scope_id (basestring): (Optional) The unique scope identifier. If no scope id is specified, a random `uuid.UUID` is used to create a new scope id. Returns ------- An `sunnydi.ioc.InjectionScope` """ injector = get_injector() return injector.scope(scope_id)
[docs]def get_injector(name=DEFAULT_INJECTOR): """ Get the `sunnydi.ioc.Injector` registered with the specified name, the default injector if no name is specified, or None if no injector is globally registered. Parameters ---------- * name (basestring): The injector name (or `DEFAULT_INJECTOR`). Returns ------- The named `sunnydi.ioc.Injector` (or the default injector if `name` is not specified) """ injector = _injectors[name] if name in _injectors else None if injector is None: return assert isinstance(injector, Injector) return injector