Skip to main content

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.

from contextlib import suppress

import aiohttp
from aiomisc import Service
from aiomisc.service.aiohttp import AIOHTTPService

class HealthcheckService(Service):

    __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',)

    ...

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.

Files for aiomisc-dependency, version 0.1.7
Filename, size File type Python version Upload date Hashes
Filename, size aiomisc_dependency-0.1.7-py3-none-any.whl (5.3 kB) File type Wheel Python version py3 Upload date Hashes View hashes
Filename, size aiomisc_dependency-0.1.7.tar.gz (4.1 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page