refactor long running manager

Author: Andrei Neagu

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_manager.py

@@ -1,81 +1,12 @@
-import datetime
-
 from aiohttp import web
-from settings_library.rabbit import RabbitSettings
-from settings_library.redis import RedisSettings
 
-from ...long_running_tasks import lrt_api
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
-from ...long_running_tasks.models import RabbitNamespace, TaskContext
-from ...long_running_tasks.task import RedisNamespace, TasksManager
-from ...rabbitmq._client_rpc import RabbitMQRPCClient
+from ...long_running_tasks.models import TaskContext
 from ._constants import APP_LONG_RUNNING_MANAGER_KEY
 from ._request import get_task_context
 
 
 class AiohttpLongRunningManager(BaseLongRunningManager):
-    def __init__(
-        self,
-        app: web.Application,
-        stale_task_check_interval: datetime.timedelta,
-        stale_task_detect_timeout: datetime.timedelta,
-        redis_settings: RedisSettings,
-        rabbit_settings: RabbitSettings,
-        redis_namespace: RedisNamespace,
-        rabbit_namespace: RabbitNamespace,
-    ):
-        self._app = app
-        self._tasks_manager = TasksManager(
-            stale_task_check_interval=stale_task_check_interval,
-            stale_task_detect_timeout=stale_task_detect_timeout,
-            redis_settings=redis_settings,
-            redis_namespace=redis_namespace,
-        )
-        self._rabbit_namespace = rabbit_namespace
-        self.rabbit_settings = rabbit_settings
-        self._rpc_server: RabbitMQRPCClient | None = None
-        self._rpc_client: RabbitMQRPCClient | None = None
-
-    @property
-    def tasks_manager(self) -> TasksManager:
-        return self._tasks_manager
-
-    @property
-    def rpc_server(self) -> RabbitMQRPCClient:
-        assert self._rpc_server is not None  # nosec
-        return self._rpc_server
-
-    @property
-    def rpc_client(self) -> RabbitMQRPCClient:
-        assert self._rpc_client is not None  # nosec
-        return self._rpc_client
-
-    @property
-    def rabbit_namespace(self) -> RabbitNamespace:
-        return self._rabbit_namespace
-
-    async def setup(self) -> None:
-        await self._tasks_manager.setup()
-        self._rpc_server = await RabbitMQRPCClient.create(
-            client_name=f"lrt-server-{self.rabbit_namespace}",
-            settings=self.rabbit_settings,
-        )
-        self._rpc_client = await RabbitMQRPCClient.create(
-            client_name=f"lrt-client-{self.rabbit_namespace}",
-            settings=self.rabbit_settings,
-        )
-        await lrt_api.register_rabbit_routes(self)
-
-    async def teardown(self) -> None:
-        await self._tasks_manager.teardown()
-
-        if self._rpc_server is not None:
-            await self._rpc_server.close()
-            self._rpc_server = None
-
-        if self._rpc_client is not None:
-            await self._rpc_client.close()
-            self._rpc_client = None
 
     @staticmethod
     def get_task_context(request: web.Request) -> TaskContext:

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_server.py

@@ -175,7 +175,6 @@ async def on_cleanup_ctx(app: web.Application) -> AsyncGenerator[None, None]:
         # add components to state
         app[APP_LONG_RUNNING_MANAGER_KEY] = long_running_manager = (
             AiohttpLongRunningManager(
-                app=app,
                 stale_task_check_interval=stale_task_check_interval,
                 stale_task_detect_timeout=stale_task_detect_timeout,
                 redis_settings=redis_settings,

packages/service-library/src/servicelib/fastapi/long_running_tasks/_manager.py

@@ -1,76 +1,5 @@
-import datetime
-
-from fastapi import FastAPI
-from settings_library.rabbit import RabbitSettings
-from settings_library.redis import RedisSettings
-
-from ...long_running_tasks import lrt_api
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
-from ...long_running_tasks.models import RabbitNamespace
-from ...long_running_tasks.task import RedisNamespace, TasksManager
-from ...rabbitmq._client_rpc import RabbitMQRPCClient
 
 
 class FastAPILongRunningManager(BaseLongRunningManager):
-    def __init__(
-        self,
-        app: FastAPI,
-        stale_task_check_interval: datetime.timedelta,
-        stale_task_detect_timeout: datetime.timedelta,
-        redis_settings: RedisSettings,
-        rabbit_settings: RabbitSettings,
-        redis_namespace: RedisNamespace,
-        rabbit_namespace: RabbitNamespace,
-    ):
-        self._app = app
-        self._tasks_manager = TasksManager(
-            stale_task_check_interval=stale_task_check_interval,
-            stale_task_detect_timeout=stale_task_detect_timeout,
-            redis_settings=redis_settings,
-            redis_namespace=redis_namespace,
-        )
-        self._rabbit_namespace = rabbit_namespace
-        self.rabbit_settings = rabbit_settings
-        self._rpc_server: RabbitMQRPCClient | None = None
-        self._rpc_client: RabbitMQRPCClient | None = None
-
-    @property
-    def tasks_manager(self) -> TasksManager:
-        return self._tasks_manager
-
-    @property
-    def rpc_server(self) -> RabbitMQRPCClient:
-        assert self._rpc_server is not None  # nosec
-        return self._rpc_server
-
-    @property
-    def rpc_client(self) -> RabbitMQRPCClient:
-        assert self._rpc_client is not None  # nosec
-        return self._rpc_client
-
-    @property
-    def rabbit_namespace(self) -> str:
-        return self._rabbit_namespace
-
-    async def setup(self) -> None:
-        await self._tasks_manager.setup()
-        self._rpc_server = await RabbitMQRPCClient.create(
-            client_name=f"lrt-server-{self.rabbit_namespace}",
-            settings=self.rabbit_settings,
-        )
-        self._rpc_client = await RabbitMQRPCClient.create(
-            client_name=f"lrt-client-{self.rabbit_namespace}",
-            settings=self.rabbit_settings,
-        )
-        await lrt_api.register_rabbit_routes(self)
-
-    async def teardown(self) -> None:
-        await self._tasks_manager.teardown()
-
-        if self._rpc_server is not None:
-            await self._rpc_server.close()
-            self._rpc_server = None
-
-        if self._rpc_client is not None:
-            await self._rpc_client.close()
-            self._rpc_client = None
+    pass

packages/service-library/src/servicelib/fastapi/long_running_tasks/_server.py

@@ -46,7 +46,6 @@ async def on_startup() -> None:
         # add components to state
         app.state.long_running_manager = long_running_manager = (
             FastAPILongRunningManager(
-                app=app,
                 stale_task_check_interval=stale_task_check_interval,
                 stale_task_detect_timeout=stale_task_detect_timeout,
                 redis_settings=redis_settings,

packages/service-library/src/servicelib/long_running_tasks/_rabbit/lrt_server.py

@@ -1,4 +1,3 @@
-import logging
 import traceback
 from typing import Any
 
@@ -10,8 +9,6 @@
 from ..task import RegisteredTaskName
 from ._models import RPCErrorResponse
 
-_logger = logging.getLogger(__name__)
-
 router = RPCRouter()
 
 

packages/service-library/src/servicelib/long_running_tasks/base_long_running_manager.py

@@ -1,34 +1,83 @@
-from abc import ABC, abstractmethod
+import datetime
+
+from settings_library.rabbit import RabbitSettings
+from settings_library.redis import RedisSettings
 
 from ..rabbitmq._client_rpc import RabbitMQRPCClient
+from ._rabbit.namespace import get_namespace
 from .models import RabbitNamespace
-from .task import TasksManager
+from .task import RedisNamespace, TasksManager
 
 
-class BaseLongRunningManager(ABC):
+class BaseLongRunningManager:
     """
     Provides a commond inteface for aiohttp and fastapi services
     """
 
+    def __init__(
+        self,
+        stale_task_check_interval: datetime.timedelta,
+        stale_task_detect_timeout: datetime.timedelta,
+        redis_settings: RedisSettings,
+        rabbit_settings: RabbitSettings,
+        redis_namespace: RedisNamespace,
+        rabbit_namespace: RabbitNamespace,
+    ):
+        self._tasks_manager = TasksManager(
+            stale_task_check_interval=stale_task_check_interval,
+            stale_task_detect_timeout=stale_task_detect_timeout,
+            redis_settings=redis_settings,
+            redis_namespace=redis_namespace,
+        )
+        self._rabbit_namespace = rabbit_namespace
+        self.rabbit_settings = rabbit_settings
+        self._rpc_server: RabbitMQRPCClient | None = None
+        self._rpc_client: RabbitMQRPCClient | None = None
+
     @property
-    @abstractmethod
     def tasks_manager(self) -> TasksManager:
-        pass
+        return self._tasks_manager
 
     @property
-    @abstractmethod
     def rpc_server(self) -> RabbitMQRPCClient:
-        pass
+        assert self._rpc_server is not None  # nosec
+        return self._rpc_server
+
+    @property
+    def rpc_client(self) -> RabbitMQRPCClient:
+        assert self._rpc_client is not None  # nosec
+        return self._rpc_client
 
     @property
-    @abstractmethod
     def rabbit_namespace(self) -> RabbitNamespace:
-        pass
+        return self._rabbit_namespace
 
-    @abstractmethod
     async def setup(self) -> None:
-        pass
+        await self._tasks_manager.setup()
+        self._rpc_server = await RabbitMQRPCClient.create(
+            client_name=f"lrt-server-{self.rabbit_namespace}",
+            settings=self.rabbit_settings,
+        )
+        self._rpc_client = await RabbitMQRPCClient.create(
+            client_name=f"lrt-client-{self.rabbit_namespace}",
+            settings=self.rabbit_settings,
+        )
+
+        from ._rabbit.lrt_server import router
+
+        await self.rpc_server.register_router(
+            router,
+            get_namespace(self.rabbit_namespace),
+            self,
+        )
 
-    @abstractmethod
     async def teardown(self) -> None:
-        pass
+        await self._tasks_manager.teardown()
+
+        if self._rpc_server is not None:
+            await self._rpc_server.close()
+            self._rpc_server = None
+
+        if self._rpc_client is not None:
+            await self._rpc_client.close()
+            self._rpc_client = None

packages/service-library/src/servicelib/long_running_tasks/lrt_api.py

@@ -1,16 +1,12 @@
-import logging
 from typing import Any
 
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 
-from ._rabbit import lrt_client, lrt_server
-from ._rabbit.namespace import get_namespace
+from ._rabbit import lrt_client
 from .base_long_running_manager import BaseLongRunningManager
 from .models import TaskBase, TaskContext, TaskId, TaskStatus
 from .task import RegisteredTaskName
 
-_logger = logging.getLogger(__name__)
-
 
 async def start_task(
     rabbitmq_rpc_client: RabbitMQRPCClient,
@@ -115,12 +111,3 @@ async def remove_task(
         task_id=task_id,
         task_context=task_context,
     )
-
-
-async def register_rabbit_routes(long_running_manager: BaseLongRunningManager) -> None:
-    rpc_server = long_running_manager.rpc_server
-    await rpc_server.register_router(
-        lrt_server.router,
-        get_namespace(long_running_manager.rabbit_namespace),
-        long_running_manager,
-    )