Skip to main content

Knighted

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"

Release files for knighted 0.3

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for knighted 0.3
File Size Uploaded
knighted-0.3.tar.gz 20.4 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for knighted 0.3
File Interpreter ABI Platform
knighted-0.3-py3-none-any.whl Python 3 none any Details

Total release size: 30.9 kB

Release files / knighted-0.3.tar.gz

Download URL knighted-0.3.tar.gz
Size 20.4 kB
Tags Source
SHA-256 checksum
How to use checksums
f9b4eac495ea925d8b2a54122a3c76dae8a5ba9f3458425b2dcc89deb6d9020f
BLAKE2b-256 checksum
How to use checksums
899c6a414d518360ade9a7f96f6f8fe8b4a566d7b681ab9099983f29a9ca6f42
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/1.12.1 pkginfo/1.4.2 requests/2.21.0 setuptools/40.6.3 requests-toolbelt/0.8.0 tqdm/4.28.1 CPython/3.7.1

Release files / knighted-0.3-py3-none-any.whl

Download URL knighted-0.3-py3-none-any.whl
Size 10.5 kB
Tags Python 3
SHA-256 checksum
How to use checksums
c9d6eefc6dc5e8636c7c95e5e39e6bb6897451cdb15e1ec4b3fc1a235c341495
BLAKE2b-256 checksum
How to use checksums
6e131d210a74774723e55f2395e59e94a10a5c5e05ac73a279668b83da969b81
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/1.12.1 pkginfo/1.4.2 requests/2.21.0 setuptools/40.6.3 requests-toolbelt/0.8.0 tqdm/4.28.1 CPython/3.7.1
Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page