Skip to main content

Asyncio plugins, components, dependency injection and configs

Project description

This is heavily inspired by Pyramid and my daily needs to fastly create and maintain microservice like applications.

a plugin mechanic

  • plugin may depend on other plugins

  • plugins yield tasks to run

  • a context registry serves as a store for application components created by plugins

  • a dependency injection creates intermediate components

  • a config source is mapped to plugin specific configuration and may be fully overridden by environment vars

  • structlog boilerplate for json/tty logging

  • fork the process and share bound sockets

  • pytest fixtures to reduce testing boilerplate

You bootstrap like following:

from buvar import plugin

# some.module.with.prepare
from buvar import context, plugin

class Foo:

async def task():

async def server():
    my_component = context.get(Foo)
    await asyncio.Future()

# there is also plugin.Teardown and plugin.Cancel
async def prepare(load: plugin.Loader):
    await load('.another.plugin')

    # create some long lasting components
    my_component = context.add(Foo())

    # you may run simple tasks
    yield task()

    # you may run server tasks
    yield server()

a components and dependency injection solution

Dependency injection relies on registered adapters, which may be a function, a method, a class, a classmethod or a generic classmthod.

Dependencies are looked up in components or may be provided via kwargs.

from buvar import di

class Bar:

class Foo:
    def __init__(self, bar: Bar = None): = bar

    async def adapt(cls, baz: str) -> Foo:
        return Foo()

async def adapt(bar: Bar) -> Foo
    foo = Foo(bar)
    return foo

async def task():
    foo = await di.nject(Foo, baz="baz")
    assert is None

    bar = Bar()
    foo = await di.nject(Foo, bar=bar)
    assert is bar

async def prepare():

    yield task()

a config source

buvar.config.ConfigSource is just a dict, which merges arbitrary dicts into one. It serves as the single source of truth for application variability.

You can load a section of config values into your custom attrs class instance. ConfigSource will override values by environment variables if present.


log_level = "DEBUG"
show_warnings = "yes"

some = "value"
export APP_FOOBAR_SOME=thing
import attr
import toml

from buvar import config

class GeneralConfig:
    log_level: str = "INFO"
    show_warnings: bool = config.bool_var(False)

class FoobarConfig:
   some: str

source = config.ConfigSource(toml.load('config.toml'), env_prefix="APP")

general_config = source.load(GeneralConfig)
assert general_config == GeneralConfig(log_level="DEBUG", show_warnings=True)

foobar_config = source.load(FoobarConfig, 'foobar')
assert foobar_config.some == "thing"

There is a shortcut to the above approach provided by buvar.config.Config, which requires to be subclassed from it with a distinct section attribute. If one adds a buvar.config.ConfigSource component, he will receive the mapped config in one call.

from buvar import config, plugin

class GeneralConfig(config.Config):
    log_level: str = "INFO"
    show_warnings: bool = config.bool_var(False)

class FoobarConfig(config.Config, section="foobar"):
    some: str

async def prepare(load: plugin.Loader):
    # this would by typically placed in the main CLI entry point
    source = context.add(config.ConfigSource(toml.load('config.toml'), env_prefix="APP"))

    # to provide the adapter to di, which could also be done in the main entry point
    await load(config)
    foobar_config = await di.nject(FoobarConfig)

a structlog

Just structlog boilerplate.

import sys

from buvar import log

log_config = log.LogConfig(tty=sys.stdout.isatty(), level="DEBUG")

forked process and shared sockets

You may fork your process and bind and share sockets, to leverage available CPUs e.g. for serving an aiohttp microservice.

Signals like INT, TERM, HUP are forwarded to the child processes.

import aiohttp.web
from buvar import fork, plugin, di, context
from buvar_aiohttp import AioHttpConfig

async def hello(request):
    return aiohttp.web.Response(body=b"Hello, world")

async def prepare_aiohttp(load: plugin.Loader):
    await load("buvar_aiohttp")

    app = await di.nject(aiohttp.web.Application)
    app.router.add_route("GET", "/", hello)

context.add(AioHttpConfig(host="", port=5678))

fork.stage(prepare_aiohttp, forks=0, sockets=["tcp://:5678"])


There are a couple of pytest fixtures provided to get your context into a reasonable state:


A dict with arbitrary application settings.


The basic context staging operates on.


The actual stage processing all plugins.


The loader to add plugins to the stage.


The context all plugins share, when they are prepared.

Following markers may be applied to a test

buvar_plugins(plugin, ...)

Load all plugins into plugin context.

import pytest

async def prepare():
    from buvar import context


async def test_wrapped_stage_context():
    from buvar import context, plugin

    assert context.get(str) == "foobar"
    assert context.get(plugin.Cancel)

async def test_wrapped_stage_context_load(buvar_load):
    from buvar import context, plugin

    await buvar_load(prepare)
    assert context.get(str) == "foobar"
    assert context.get(plugin.Cancel)

Project details

Download files

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

Source Distribution

buvar-0.43.0.tar.gz (134.0 kB view hashes)

Uploaded source

Built Distributions

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page