Last active
November 20, 2015 13:26
-
-
Save popravich/c2382be9bf20e4da1816 to your computer and use it in GitHub Desktop.
Injections extension for propagating deps into standalone functions
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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 = injections.dependencies(source) | |
| 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].type), ( | |
| "Dependencies missmatch {!r}({!r}) {!r}" | |
| .format(dep_name, dep_type, deps[dep_name].type)) | |
| 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)) | |
| 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_ = {} | |
| exec(co, {'source': source, fun.__name__: fun}, locals_) | |
| 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