Injections extension for propagating deps into standalone functions

README.rst

Dependency injections function decorator

Small extension for injections library.

Main idea for this kind of injections is to get rid from something like this:

# my_package/service.py

from very.black.magic.proxing import Local

proxy = Local()
service = proxy('service')

def init_service(config):
    proxy.service = Service(config)

# my_package/handler.py

from .service import service

def handle_request():
    service.do_something()

...and replace it with this:

# my_package/deps.py
import injections as di

@di.has
class CommonDeps:
    service = di.depends(Service)

common_deps_provider = CommonDeps()
# at this point dependencies has not been injected

# my_package/handler.py

from .service import Service
from .deps import common_deps_provider

@inject_from(common_deps_provider)
def handle_request(service: Service):
    service.do_something()

# and somewhere in main:

inj = inections.Container()
inj['service'] = Service(config)
inj.inject(common_deps_provider)

func_inject.py

import inspect
import injections
import functools
import textwrap


def inject_from(source):
    """Decorator for injecting dependencies into standalone functions.

    source argument is an instance of any object marked with ``injections.has``

    decorated function must use annotations for arguments which must be
    replaced by dependency.

    Usage::

    >>> @injections.has
    ... class SomeDeps:
    ...     redis = injections.depends(Redis)

    >>> inst = SomeDeps()

    >>> @inject_from(inst)
    ... def get_key(key, redis: Redis):
    ...     return redis.get(key)

    >>> print(get_key('foo'))
    """

    deps = {d.name: d.type for d in injections.dependencies(source).values()}
    assert deps, "Object {!r} has no dependencies".format(source)

    def wrapper(fun):
        sig = inspect.signature(fun)
        params = sig.parameters.copy()
        allowed_kinds = (inspect.Parameter.POSITIONAL_OR_KEYWORD,
                         inspect.Parameter.KEYWORD_ONLY)
        fun_args = []

        def push_fun_arg(param):
            if param.kind is param.KEYWORD_ONLY:
                fun_args.append('{0}={0}'.format(param.name))
            else:
                fun_args.append(str(param.replace(annotation=param.empty,
                                                  default=param.empty)))

        for name, param in list(params.items()):
            ann = param.annotation
            if ann is param.empty:
                push_fun_arg(param)
                continue
            if isinstance(ann, injections.Dependency):
                dep_type = ann.type
                dep_name = ann.name or name
            elif isinstance(ann, type) and name in deps:
                dep_type = ann
                dep_name = name
            else:
                push_fun_arg(param)
                continue
            assert param.kind in allowed_kinds, param
            assert param.default is param.empty, (
                "Parameter {!r} has default value".format(name))
            assert dep_name in deps, (
                "Unknown dependency {!r}({!r}) {!r}"
                .format(dep_name, dep_type, deps))
            assert issubclass(dep_type, deps[dep_name]), (
                "Dependencies missmatch {!r}({!r}) {!r}"
                .format(dep_name, dep_type, deps[dep_name]))
            params.pop(name)
            if param.kind is param.KEYWORD_ONLY:
                fun_args.append('{}=source.{}'.format(name, dep_name))
            else:
                fun_args.append('source.{}'.format(dep_name))
        # FIXME: wrapper signature may contain custom annotations
        #        which should be visible to exec
        code = """
            def wrapper{sig}:
                return {fun}({fun_args})
            """
        code = code.format(sig=sig.replace(parameters=params.values()),
                           fun=fun.__name__, fun_args=', '.join(fun_args))
        code = textwrap.dedent(code)
        co = compile(code, '<string>', 'exec')
        locals_ = {}
        try:
            exec(co, {'source': source, fun.__name__: fun}, locals_)
        except NameError:
            raise AssertionError(
                "Possibly some dependencies were not resolved:\n{!s}"
                .format(code))
        wrapper = locals_['wrapper']
        return functools.wraps(fun)(wrapper)

    return wrapper