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
plugin.stage("some.module.with.prepare")# some.module.with.prepare
from buvar import context, plugin
class Foo:
...
async def task():
asyncio.sleep(1)
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:
pass
class Foo:
def __init__(self, bar: Bar = None):
self.bar = bar
@classmethod
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 foo.bar is None
bar = Bar()
foo = await di.nject(Foo, bar=bar)
assert foo.bar is bar
async def prepare():
di.register(Foo.adapt)
di.register(adapt)
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.
config.toml
log_level = "DEBUG"
show_warnings = "yes"
[foobar]
some = "value"export APP_FOOBAR_SOME=thingimport attr
import toml
from buvar import config
@attr.s(auto_attribs=True)
class GeneralConfig:
log_level: str = "INFO"
show_warnings: bool = config.bool_var(False)
@attr.s(auto_attribs=True)
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
@attr.s(auto_attribs=True)
class GeneralConfig(config.Config):
log_level: str = "INFO"
show_warnings: bool = config.bool_var(False)
@attr.s(auto_attribs=True)
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")
log_config.setup()pytest
There are a couple of pytest fixtures provided to get your context into a reasonable state:
buvar_config_sourceA
dictwith arbitrary application settings.buvar_contextThe basic context staging operates on.
buvar_stageThe actual stage processing all plugins.
buvar_loadThe loader to add plugins to the stage.
buvar_plugin_contextThe 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
context.add("foobar")
@pytest.mark.asyncio
@pytest.mark.buvar_plugins("tests.test_testing")
async def test_wrapped_stage_context():
from buvar import context, plugin
assert context.get(str) == "foobar"
assert context.get(plugin.Cancel)
@pytest.mark.asyncio
@pytest.mark.buvar_plugins()
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)Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
File details
Details for the file buvar-0.43.6.tar.gz.
File metadata
- Download URL: buvar-0.43.6.tar.gz
- Upload date:
- Size: 33.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via: twine/5.0.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | eb8b1a41c5afec4ee1846041f2032ea5f41e2762261e4f7d6e4e56ccd5567429 |
|
| MD5 | 7ea897e7338d6ff27b5f27e382601c81 |
|
| BLAKE2b-256 | 53ec023c81758da28ee26516fbf92fd324ae066a0dc950eac5c22e78fc60a383 |
