Merge branch 'release/0.1.1'

Author: s3rius

pyproject.toml

@@ -3,7 +3,7 @@ name = "taskiq-aiohttp"
 description = "Taskiq integration with AioHTTP framework"
 authors = ["Taskiq team <[email protected]>"]
 maintainers = ["Taskiq team <[email protected]>"]
-version = "0.1.0"
+version = "0.1.1"
 readme = "README.md"
 license = "LICENSE"
 classifiers = [

taskiq_aiohttp/initializer.py

@@ -1,6 +1,6 @@
 import asyncio
 import inspect
-from typing import Awaitable, Callable
+from typing import Any, Awaitable, Callable
 
 import yarl
 from aiohttp import web
@@ -16,6 +16,7 @@
 def startup_event_generator(
     broker: AsyncBroker,
     app_path: str,
+    app: Any,
 ) -> Callable[[TaskiqState], Awaitable[None]]:
     """
     Creates an event to run on broker's startup.
@@ -27,26 +28,27 @@ def startup_event_generator(
     act the same as the real application.
 
     :param broker: current broker.
-    :param app_path: string with a path to an application or a factory.
+    :param app_path: path to the application.
+    :param app: current application or a fractory.
 
     :returns: a function that is called on startup.
     """
 
     async def startup(state: TaskiqState) -> None:
         loop = asyncio.get_event_loop()
 
-        app = import_object(app_path)
+        local_app = app
 
-        if not isinstance(app, web.Application):
-            app = app()
+        if not isinstance(local_app, web.Application):
+            local_app = local_app()
 
-        if inspect.iscoroutine(app):
-            app = await app
+        if inspect.iscoroutine(local_app):
+            local_app = await local_app
 
-        if not isinstance(app, web.Application):
-            raise ValueError(f"'{app_path}' is not an AioHTTP application.")
+        if not isinstance(local_app, web.Application):
+            raise ValueError(f"{app_path} is not an AioHTTP application.")
 
-        handler = RequestHandler(app._make_handler(), loop=loop)
+        handler = RequestHandler(local_app._make_handler(), loop=loop)
         handler.transport = asyncio.Transport()
         request = web.Request(
             RawRequestMessage(
@@ -76,19 +78,19 @@ async def startup(state: TaskiqState) -> None:
             match_dict={},
             route=SystemRoute(web.HTTPBadRequest()),
         )
-        request._match_info._apps = app._subapps
-        request._match_info._current_app = app
+        request._match_info._apps = local_app._subapps
+        request._match_info._current_app = local_app
 
         broker.add_dependency_context(
             {
-                web.Application: app,
+                web.Application: local_app,
                 web.Request: request,
             },
         )
 
-        state.aiohttp_app = app
-        app.router._resources = []
-        await app.startup()
+        state.aiohttp_app = local_app
+        local_app.router._resources = []
+        await local_app.startup()
 
     return startup
 
@@ -121,9 +123,11 @@ def init(broker: AsyncBroker, app_path: str) -> None:
     if not broker.is_worker_process:
         return
 
+    app = import_object(app_path)
+
     broker.add_event_handler(
         TaskiqEvents.WORKER_STARTUP,
-        startup_event_generator(broker, app_path),
+        startup_event_generator(broker, app_path, app),
     )
     broker.add_event_handler(
         TaskiqEvents.WORKER_SHUTDOWN,