General purpose asyncio service loader and dependency injector
Project description
objective
Basically I want app similar, what Pyramid provides, but for asyncio and for a all kinds of services.
Have a plugin system, which runs code not on import time, but on run time. So you can test and mock your code.
Have a component registry to hold certain state of your application.
Have a simple way to configure your application via OS environment.
Have always a structlog.
a use case
src/app/__main__.py
"""Main entry point to run the server."""
import asyncio
import os
import sys
import typing
import attr
import toml
import structlog
from buvar import components, config, log, plugin
@attr.s(auto_attribs=True)
class GeneralConfig:
"""Simple config."""
log_level: str = "INFO"
plugins: typing.Set[str] = set()
loop = asyncio.get_event_loop()
user_config = toml.load(
os.environ.get("USER_CONFIG", os.path.dirname(__file__) + "/user_config.toml")
)
cmps = components.Components()
source = cmps.add(config.ConfigSource(user_config, env_prefix="APP"))
general_config = cmps.add(source.load(GeneralConfig))
log.setup_logging(sys.stdout.isatty(), general_config.log_level)
sl = structlog.get_logger()
sl.info("Starting process", pid=os.getpid())
sl.debug("Config used", **source)
plugin.run(*general_config.plugins, components=cmps, loop=loop)src/app/user_config.toml
log_level = "DEBUG"
plugins = ['app']
[aiohttp]
host = "127.0.0.1"
port = 5000src/app/__init__.py
import aiohttp.web
import structlog
from buvar import context
log = structlog.get_logger()
routes = aiohttp.web.RouteTableDef()
@routes.get("/")
async def index(request):
return aiohttp.web.json_response({"hello": "world"})
async def plugin(include):
await include(".server")
app = context.get(aiohttp.web.Application)
app.add_routes(routes)src/app/server.py
"""Create a aiohttp server task and provide the application via context."""
import aiohttp.web
import attr
import structlog
from buvar import config, context
sl = structlog.get_logger()
@attr.s(auto_attribs=True)
class AioHttpConfig:
host: str = "0.0.0.0"
port: int = 8080
class AccessLogger(aiohttp.abc.AbstractAccessLogger): # noqa: R0903
def log(self, request, response, time): # noqa: R0201
sl.info(
"Access",
remote=request.remote,
method=request.method,
path=request.path,
time=time,
status=response.status,
)
async def plugin():
config_source = context.get(config.ConfigSource)
aiohttp_config = config_source.load(AioHttpConfig, "aiohttp")
aiohttp_app = context.add(
aiohttp.web.Application(middlewares=[aiohttp.web.normalize_path_middleware()])
)
sl.info("Running web server", host=aiohttp_config.host, port=aiohttp_config.port)
# return server task
yield aiohttp.web._run_app( # noqa: W0212
aiohttp_app, host=aiohttp_config.host, port=aiohttp_config.port, print=None
)cd src
APP_AIOHTTP_HOST=0.0.0.0 APP_AIOHTTP_PORT=8080 python -m app2019-07-09T20:52:40.979551Z [info ] Starting process [__main__] pid=13158
2019-07-09T20:52:40.979753Z [debug ] Config used [__main__] aiohttp={'host': '127.0.0.1', 'port': 5000} log_level=DEBUG pid=13158 plugins=['app']
2019-07-09T20:52:40.980269Z [debug ] Using selector: EpollSelector [asyncio] pid=13158
2019-07-09T20:52:40.981489Z [info ] Running web server [app.server] host=0.0.0.0 pid=13158 port=8080Release 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
buvar-0.9.1.tar.gz
(24.5 kB
view details)
File details
Details for the file buvar-0.9.1.tar.gz.
File metadata
- Download URL: buvar-0.9.1.tar.gz
- Upload date:
- Size: 24.5 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/1.13.0 pkginfo/1.5.0.1 requests/2.22.0 setuptools/41.0.1 requests-toolbelt/0.9.1 tqdm/4.32.1 CPython/3.7.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | dfcd9af51b627a33dd74814fd676d586ccf9aa8cf4ebb9e02bfaf0b286bac2f9 |
|
| MD5 | 65b7ffd737c9fafe1b57afb457e54212 |
|
| BLAKE2b-256 | 18bd02dba78e1d76776295cfe5d7364b4fad081db854d3aedb9ad556a8468b3f |
