Skip to content

Instantly share code, notes, and snippets.

@popravich
Last active November 20, 2015 13:26
Show Gist options
  • Save popravich/c2382be9bf20e4da1816 to your computer and use it in GitHub Desktop.
Save popravich/c2382be9bf20e4da1816 to your computer and use it in GitHub Desktop.
Injections extension for propagating deps into standalone functions

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)
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment