Commit a2a8eeb8 authored by xa's avatar xa

it fills the gaps

parent 6cc0b138
......@@ -58,6 +58,28 @@ The :func:`func` can be a function or an awaitable. These 2 examples works the s
'bar': 'I am bar'}
When applied with some arguments, placeholders just fills the gaps::
@annotate('foo', 'bar')
def fun(foo, bar):
return {'foo': foo,
'bar': bar}
assert (yield from services.apply(fun, foo="yes")) == {'foo': 'yes',
'bar': 'I am bar'}
@annotate('foo', 'bar')
async def awaitable_fun(foo, bar):
return {'foo': foo,
'bar': bar}
assert (yield from services.apply(awaitable_fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
Factories also can be either sync or awaitable::
......@@ -2,10 +2,12 @@ from __future__ import annotations
import asyncio
import logging
from abc import ABCMeta
from collections import defaultdict, namedtuple, OrderedDict
from collections import defaultdict, OrderedDict
from itertools import chain
from weakref import WeakKeyDictionary
from functools import wraps
from inspect import signature
from cached_property import cached_property
logger = logging.getLogger("knighted")
......@@ -124,28 +126,39 @@ class Injector(metaclass=ABCMeta):
def wrapper(*args, **kwargs):
if func in ANNOTATIONS:
annotated = ANNOTATIONS[func]
service_args, service_kwargs = [], {}
for note in annotated.pos_notes:
service = yield from self.get(note)
for key, note in annotated.kw_notes.items():
service = yield from self.get(note)
service_kwargs[key] = service
result = func(*service_args, **service_kwargs)
annotation = ANNOTATIONS[func]
given = annotation.given(*args, **kwargs)
to_load = {}
for key, note in annotation.marked.items():
if key not in given:
to_load[key] = asyncio.create_task(self.get(note))
for key, fut in to_load.items():
to_load[key] = yield from fut
result = func(*args, **kwargs)
if asyncio.iscoroutine(result):
result = yield from result
return result
logger.warn('%r is not annoted' % func)
logger.warning('%r is not annoted', func)
return func(*args, **kwargs)
return wrapper
ANNOTATIONS: WeakKeyDictionary[str, "Annotation"] = WeakKeyDictionary()
class Annotation:
def __init__(self, pos_notes, kw_notes, func):
self.pos_notes = pos_notes
self.kw_notes = kw_notes
self.bind_partial = signature(func).bind_partial
Annotation = namedtuple('Annotation', 'pos_notes kw_notes')
def marked(self):
return self.bind_partial(*self.pos_notes, **self.kw_notes).arguments
def given(self, *args, **kwargs):
return list(self.bind_partial(*args, **kwargs).arguments)
ANNOTATIONS: WeakKeyDictionary[str, Annotation] = WeakKeyDictionary()
def close_reaction(obj):
......@@ -155,7 +168,7 @@ def close_reaction(obj):
def annotate(*args, **kwargs):
def decorate(func):
ANNOTATIONS[func] = Annotation(args, kwargs)
ANNOTATIONS[func] = Annotation(args, kwargs, func)
return func
for arg in chain(args, kwargs.values()):
......@@ -9,7 +9,9 @@ setup(
description='inject dependencies',
':python_version=="3.3"': ['asyncio'],
......@@ -36,12 +36,54 @@ async def test_instance_factory():
assert (await services.get('foo')) == 'I am foo'
assert (await services.get('bar')) == 'I am bar'
assert (await services.get('all')) == ['I am foo', 'I am bar']
# synchroneous
assert (await services.apply(fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
# asynchroneous
assert (await services.apply(awaitable_fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
async def test_fill_the_gaps():
class MyInjector(Injector):
services = MyInjector()
def foo_factory():
return 'I am foo'
async def bar_factory():
return 'I am bar'
async def together_factory():
foo = await services.get('foo:1')
bar = await services.get('bar:2')
return [foo, bar]
@annotate('foo:1', 'bar:2')
def fun(foo, bar):
return {'foo': foo,
'bar': bar}
assert (await services.get('foo:1')) == 'I am foo'
assert (await services.get('bar:2')) == 'I am bar'
assert (await services.apply(fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
assert (await services.apply(fun, "obviously")) == {'foo': 'obviously',
'bar': 'I am bar'}
assert (await services.apply(fun, "obviously", "not")) == {'foo': 'obviously',
'bar': 'not'}
assert (await services.apply(fun, foo="sometimes")) == {'foo': 'sometimes',
'bar': 'I am bar'}
assert (await services.apply(fun, bar="I feel")) == {'foo': 'I am foo',
'bar': 'I feel'}
async def test_partial():
class MyInjector(Injector):
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment