Skip to main content

inject dependencies

Project description

Knighted, is heavily inspired by jeni and works only with asyncio. It allows to described dependencies, and inject them later.

For example:

from knighted import annotation, Injector

class MyInjector(Injector):
    pass

services = MyInjector()

@services.factory('foo')
def foo_factory():
    return 'I am foo'

@services.factory('bar')
def bar_factory():
    return 'I am bar'

@services.factory('all')
def together_factory():
    foo = yield from services.get('foo')
    bar = yield from services.get('bar')
    return [foo, bar]

@annotate('foo', 'bar')
def fun(foo, bar):
    return {'foo': foo,
            'bar': bar}

assert (yield from services.apply(fun)) == {'foo': 'I am foo',
                                            'bar': 'I am bar'}

The func() can be a function or an awaitable. These 2 examples works the same:

@annotate('foo', 'bar')
def sync_fun(foo, bar):
    return {'foo': foo,
            'bar': bar}

assert (yield from services.apply(sync_fun)) == {'foo': 'I am foo',
                                                 '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'}

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:

@services.factory('bar:sync')
def bar_factory():
    return 'I am bar'

@services.factory('bar:awaitable')
async def bar_factory():
    return 'I am bar'

Services are by default singleton, but they can also be instantiated at every call:

@services.factory('bar', singleton=True)
def bar_factory():
    return time()

result1 = await services.get('bar')
sleep(.1)
result2 = await services.get('bar')
assert result1 == result2

# cache can be resetted

services.refresh("bar")
result3 = await services.get('bar')
assert result3 != result2

Singleton mode can be disabled per service:

@services.factory('baz', singleton=False)
def baz_factory():
    return time()

result1 = await services.get('baz')
sleep(.1)
result2 = await services.get('baz')
assert result1 != result2

Current services are automatically exposed inside functions:

def func():
    return current_injector()

assert func() is None
assert (await services.apply(func)) is services

Implementation

annotate(*args, **kwargs) annotate a func with service names.

coroutine Injector.factory(name) declare a service factory

coroutine Injector.get(name) return the service instance

coroutine Injector.apply(func, *args, **kwargs) call the annoted callable with the mounted service.

coroutine Injector.partial(func) prepare an annoted func with later services.

coroutine Injector.close() clear all cached services., and call all deferred close().

coroutine Injector.close.register(obj) defers yield from obj.close() when Injector.close() is called.

Factories don’t need to be fully qualified. For example:

@services.factory('prefix')
def foo_factory(bar):
    return 'I am foo and ' + bar

assert (yield from services.get('prefix:baz')) == 'I am foo and baz'
assert (yield from services.get('prefix:qux')) == 'I am foo and qux'

Closing callback can be registered:

class Foo:
    def close(self):
        self.closed = True
foo = Foo()
services.close.register(foo)
services.close()
assert foo.closed == True

Annotated functions can be rendered partially:

@annotate('foo', 'bar')
def fun(foo, bar):
    return {'foo': foo,
            'bar': bar}

partial = services.partial(fun)
assert (yield from partial()) == {'foo': 'I am foo',
                                  'bar': 'I am bar'}

Injector has a mapping interface, which allows to register arbitrary values:

services["foo"] = "yes"
assert await services["foo"] == "yes"

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date
knighted-0.3-py3-none-any.whl (10.5 kB) Copy SHA256 hash SHA256 Wheel py3
knighted-0.3.tar.gz (20.4 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page