aiomisc-dependency - dependency injection in aiomisc
Project description
Dependency injection plugin for aiomisc built with aiodine library and support pytest fixture style dependency injection.
Installation
Installing from pypi:
pip3 install aiomisc aiomisc-dependency
How to use
Register dependency
To register dependency you can use aiomisc_dependency.dependency decorator.
from aiomisc_dependency import dependency
@dependency
async def pg_engine():
pg_engine = await create_engine(dsn=pg_url)
yield pg_engine
pg_engine.close()
await pg_engine.wait_closed()
As you can see dependency can be async generator function. Code after yield will be executed on teardown to correctly close the dependency.
Coroutine functions, non async functions and generators are also supported.
Use dependency
To use dependency you need to add it’s name to __dependencies__ property for every service which depends on it. Specified dependencies will be injected as service’s attributes on entrypoint startup. If needed to map the dependency with a different name, then, use __dependencies_map__
from contextlib import suppress
from types import MappingProxyType
import aiohttp
from aiomisc.service.aiohttp import AIOHTTPService
class HealthcheckService(AIOHTTPService):
__dependencies__ = ('pg_engine',)
async def create_application(self):
app = aiohttp.web.Application()
app.add_routes([aiohttp.web.get('/ping', self.healthcheck_handler)])
return app
async def healthcheck_handler(self, request):
pg_status = False
with suppress(Exception):
async with self.pg_engine.acquire() as conn:
await conn.execute('SELECT 1')
pg_status = True
return aiohttp.web.json_response(
{'db': pg_status},
status=(200 if pg_status else 500),
)
class RESTService(AIOHTTPService):
__dependencies__ = ('pg_engine',)
...
class AnotherRESTService(AIOHTTPService):
__dependencies_map__ = MappingProxyType({'pg_engine': 'engine'})
...
If any required dependency won’t be found on entrypoint startup, RuntimeError will be raised.
You can set a dependency manually by adding it to kw arguments on service creation. This could be convenient in tests.
from unittest import Mock
def test_rest_service():
pg_engine_mock = Mock()
service = RESTService(pg_engine=pg_engine_mock)
...
Dependencies for dependencies
You can use dependencies as arguments for other dependencies. Arguments will injected automatically.
@dependency
async def pg_connection(pg_engine):
async with pg_engine.acquire() as conn:
yield conn
loop built-in dependency
Built-in loop dependency can be used if your dependency requires event loop instance.
import aioredis
@dependency
async def redis_pool(loop):
pool = aioredis.create_pool(redis_url, loop=loop)
yield pool
pool.close()
await pool.wait_closed()
LICENSE
MIT
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
Built Distribution
Hashes for aiomisc_dependency-0.1.20.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | ce40db771f8a02855307797af0058bbf4d978c6a66cddde8a033735509bb9262 |
|
MD5 | d2ed9cb6ab9a9864c09f14e1b28df44f |
|
BLAKE2b-256 | a3aa267661dcde9a98a5eec668f7343fb0e60c4aaf6bb53df91206bb6badcc71 |
Hashes for aiomisc_dependency-0.1.20-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 353609b871c7b69641300d9069b2033a76b1bd8806e6049696729b64b9e5b87a |
|
MD5 | 826bceb00f0fb3cf675ef184f70e210c |
|
BLAKE2b-256 | 85d5fd684456780a84be343106fd45f329ca56114e09d9df1a8e080c50e18bd0 |