aiomisc-dependency 0.1.4

Register dependency

要注册依赖项，可以使用aiomisc_dependency.dependencydecorator。

fromaiomisc_dependencyimportdependency@dependencyasyncdefpg_engine():pg_engine=awaitcreate_engine(dsn=pg_url)yieldpg_enginepg_engine.close()awaitpg_engine.wait_closed()

如您所见，依赖项可以是异步生成器函数。屈服后编码 将在拆卸时执行以正确关闭依赖项。

还支持协同路由函数、非异步函数和生成器。

Use dependency

要使用依赖项，需要将其名称添加到__dependencies__属性中 为每一个依赖它的服务。将插入指定的依赖项 作为入口点启动时的服务属性。

fromcontextlibimportsuppressimportaiohttpfromaiomiscimportServicefromaiomisc.service.aiohttpimportAIOHTTPServiceclassHealthcheckService(Service):__dependencies__=('pg_engine',)asyncdefcreate_application(self):app=aiohttp.web.Application()app.add_routes([aiohttp.web.get('/ping',self.healthcheck_handler)])returnappasyncdefhealthcheck_handler(self,request):pg_status=Falsewithsuppress(Exception):asyncwithself.pg_engine.acquire()asconn:awaitconn.execute('SELECT 1')pg_status=Truereturnaiohttp.web.json_response({'db':pg_status},status=(200ifpg_statuselse500),)classRESTService(AIOHTTPService):__dependencies__=('pg_engine',)...

如果在入口点启动时找不到任何必需的依赖项， RuntimeError将被筹集。

通过将依赖项添加到服务的kw参数中，可以手动设置依赖项 创造。这在测试中可能很方便。

fromunittestimportMockdeftest_rest_service():pg_engine_mock=Mock()service=RESTService(pg_engine=pg_engine_mock)...

Dependencies for dependencies

可以将依赖项用作其他依赖项的参数。参数将 自动注射。

@dependencyasyncdefpg_connection(pg_engine):asyncwithpg_engine.acquire()asconn:yieldconn

^{tt1}$ built-in dependency

如果依赖项需要，可以使用内置的loop依赖项 事件循环实例。

importaioredis@dependencyasyncdefredis_pool(loop):pool=aioredis.create_pool(redis_url,loop=loop)yieldpoolpool.close()awaitpool.wait_closed()