replaced TasksManager with BaseLongRunningManager in lrt_api

Author: Andrei Neagu

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

@@ -29,7 +29,7 @@ async def list_tasks(request: web.Request) -> web.Response:
                 abort_href=f"{request.app.router['cancel_and_delete_task'].url_for(task_id=t.task_id)}",
             )
             for t in await lrt_api.list_tasks(
-                long_running_manager.tasks_manager,
+                long_running_manager,
                 long_running_manager.get_task_context(request),
             )
         ]
@@ -42,7 +42,7 @@ async def get_task_status(request: web.Request) -> web.Response:
     long_running_manager = get_long_running_manager(request.app)
 
     task_status = await lrt_api.get_task_status(
-        long_running_manager.tasks_manager,
+        long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )
@@ -56,19 +56,19 @@ async def get_task_result(request: web.Request) -> web.Response | Any:
 
     # NOTE: this might raise an exception that will be catched by the _error_handlers
     return await lrt_api.get_task_result(
-        long_running_manager.tasks_manager,
+        long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )
 
 
-@routes.delete("/{task_id}", name="cancel_and_delete_task")
-async def cancel_and_delete_task(request: web.Request) -> web.Response:
+@routes.delete("/{task_id}", name="remove_task")
+async def remove_task(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(_PathParam, request)
     long_running_manager = get_long_running_manager(request.app)
 
     await lrt_api.remove_task(
-        long_running_manager.tasks_manager,
+        long_running_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
     )

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

@@ -63,7 +63,7 @@ async def start_long_running_task(
     task_id = None
     try:
         task_id = await lrt_api.start_task(
-            long_running_manager.tasks_manager,
+            long_running_manager,
             registerd_task_name,
             fire_and_forget=fire_and_forget,
             task_context=task_context,
@@ -97,9 +97,7 @@ async def start_long_running_task(
     except asyncio.CancelledError:
         # remove the task, the client was disconnected
         if task_id:
-            await lrt_api.remove_task(
-                long_running_manager.tasks_manager, task_context, task_id
-            )
+            await lrt_api.remove_task(long_running_manager, task_context, task_id)
         raise
 
 

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

@@ -29,9 +29,7 @@ async def list_tasks(
                 request.url_for("cancel_and_delete_task", task_id=t.task_id)
             ),
         )
-        for t in await lrt_api.list_tasks(
-            long_running_manager.tasks_manager, task_context={}
-        )
+        for t in await lrt_api.list_tasks(long_running_manager, task_context={})
     ]
 
 
@@ -52,7 +50,7 @@ async def get_task_status(
 ) -> TaskStatus:
     assert request  # nosec
     return await lrt_api.get_task_status(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
+        long_running_manager, task_context={}, task_id=task_id
     )
 
 
@@ -75,28 +73,26 @@ async def get_task_result(
 ) -> TaskResult | Any:
     assert request  # nosec
     return await lrt_api.get_task_result(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
+        long_running_manager, task_context={}, task_id=task_id
     )
 
 
 @router.delete(
     "/{task_id}",
-    summary="Cancel and deletes a task",
+    summary="Cancels and removes a task",
     response_model=None,
     status_code=status.HTTP_204_NO_CONTENT,
     responses={
         status.HTTP_404_NOT_FOUND: {"description": "Task does not exist"},
     },
 )
 @cancel_on_disconnect
-async def cancel_and_delete_task(
+async def remove_task(
     request: Request,
     task_id: TaskId,
     long_running_manager: Annotated[
         FastAPILongRunningManager, Depends(get_long_running_manager)
     ],
 ) -> None:
     assert request  # nosec
-    await lrt_api.remove_task(
-        long_running_manager.tasks_manager, task_context={}, task_id=task_id
-    )
+    await lrt_api.remove_task(long_running_manager, task_context={}, task_id=task_id)

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

@@ -4,15 +4,16 @@
 from common_library.error_codes import create_error_code
 
 from ..logging_errors import create_troubleshootting_log_kwargs
+from .base_long_running_manager import BaseLongRunningManager
 from .errors import TaskNotCompletedError, TaskNotFoundError
 from .models import TaskBase, TaskContext, TaskId, TaskStatus
-from .task import RegisteredTaskName, TasksManager
+from .task import RegisteredTaskName
 
 _logger = logging.getLogger(__name__)
 
 
 async def start_task(
-    tasks_manager: TasksManager,
+    long_running_manager: BaseLongRunningManager,
     registered_task_name: RegisteredTaskName,
     *,
     unique: bool = False,
@@ -46,7 +47,7 @@ async def start_task(
     Returns:
         TaskId: the task unique identifier
     """
-    return await tasks_manager.start_task(
+    return await long_running_manager.tasks_manager.start_task(
         registered_task_name,
         unique=unique,
         task_context=task_context,
@@ -57,28 +58,34 @@ async def start_task(
 
 
 async def list_tasks(
-    tasks_manager: TasksManager, task_context: TaskContext
+    long_running_manager: BaseLongRunningManager, task_context: TaskContext
 ) -> list[TaskBase]:
-    return await tasks_manager.list_tasks(with_task_context=task_context)
+    return await long_running_manager.tasks_manager.list_tasks(
+        with_task_context=task_context
+    )
 
 
 async def get_task_status(
-    tasks_manager: TasksManager, task_context: TaskContext, task_id: TaskId
+    long_running_manager: BaseLongRunningManager,
+    task_context: TaskContext,
+    task_id: TaskId,
 ) -> TaskStatus:
     """returns the status of a task"""
-    return await tasks_manager.get_task_status(
+    return await long_running_manager.tasks_manager.get_task_status(
         task_id=task_id, with_task_context=task_context
     )
 
 
 async def get_task_result(
-    tasks_manager: TasksManager, task_context: TaskContext, task_id: TaskId
+    long_running_manager: BaseLongRunningManager,
+    task_context: TaskContext,
+    task_id: TaskId,
 ) -> Any:
     try:
-        task_result = await tasks_manager.get_task_result(
+        task_result = await long_running_manager.tasks_manager.get_task_result(
             task_id, with_task_context=task_context
         )
-        await tasks_manager.remove_task(
+        await long_running_manager.tasks_manager.remove_task(
             task_id, with_task_context=task_context, reraise_errors=False
         )
         return task_result
@@ -94,14 +101,18 @@ async def get_task_result(
             ),
         )
         # the task shall be removed in this case
-        await tasks_manager.remove_task(
+        await long_running_manager.tasks_manager.remove_task(
             task_id, with_task_context=task_context, reraise_errors=False
         )
         raise
 
 
 async def remove_task(
-    tasks_manager: TasksManager, task_context: TaskContext, task_id: TaskId
+    long_running_manager: BaseLongRunningManager,
+    task_context: TaskContext,
+    task_id: TaskId,
 ) -> None:
     """cancels and removes the task"""
-    await tasks_manager.remove_task(task_id, with_task_context=task_context)
+    await long_running_manager.tasks_manager.remove_task(
+        task_id, with_task_context=task_context
+    )

packages/service-library/tests/aiohttp/long_running_tasks/test_long_running_tasks.py

@@ -127,7 +127,7 @@ async def test_workflow(
     [
         ("GET", "get_task_status"),
         ("GET", "get_task_result"),
-        ("DELETE", "cancel_and_delete_task"),
+        ("DELETE", "remove_task"),
     ],
 )
 async def test_get_task_wrong_task_id_raises_not_found(
@@ -188,7 +188,7 @@ async def test_cancel_task(
     task_id = await start_long_running_task(client)
 
     # cancel the task
-    delete_url = client.app.router["cancel_and_delete_task"].url_for(task_id=task_id)
+    delete_url = client.app.router["remove_task"].url_for(task_id=task_id)
     result = await client.delete(f"{delete_url}")
     data, error = await assert_status(result, status.HTTP_204_NO_CONTENT)
     assert not data

packages/service-library/tests/fastapi/long_running_tasks/test_long_running_tasks.py

@@ -81,7 +81,7 @@ async def create_string_list_task(
         fail: bool = False,
     ) -> TaskId:
         return await lrt_api.start_task(
-            long_running_manager.tasks_manager,
+            long_running_manager,
             _string_list_task.__name__,
             num_strings=num_strings,
             sleep_time=sleep_time,
@@ -205,7 +205,7 @@ async def test_workflow(
     [
         ("GET", "get_task_status"),
         ("GET", "get_task_result"),
-        ("DELETE", "cancel_and_delete_task"),
+        ("DELETE", "remove_task"),
     ],
 )
 async def test_get_task_wrong_task_id_raises_not_found(
@@ -254,7 +254,7 @@ async def test_cancel_task(
     task_id = await start_long_running_task(app, client)
 
     # cancel the task
-    delete_url = app.url_path_for("cancel_and_delete_task", task_id=task_id)
+    delete_url = app.url_path_for("remove_task", task_id=task_id)
     result = await client.delete(f"{delete_url}")
     assert result.status_code == status.HTTP_204_NO_CONTENT
 

packages/service-library/tests/fastapi/long_running_tasks/test_long_running_tasks_context_manager.py

@@ -71,9 +71,7 @@ async def create_task_user_defined_route(
             FastAPILongRunningManager, Depends(get_long_running_manager)
         ],
     ) -> TaskId:
-        return await lrt_api.start_task(
-            long_running_manager.tasks_manager, a_test_task.__name__
-        )
+        return await lrt_api.start_task(long_running_manager, a_test_task.__name__)
 
     @router.get("/api/failing", status_code=status.HTTP_200_OK)
     async def create_task_which_fails(
@@ -82,7 +80,7 @@ async def create_task_which_fails(
         ],
     ) -> TaskId:
         return await lrt_api.start_task(
-            long_running_manager.tasks_manager, a_failing_test_task.__name__
+            long_running_manager, a_failing_test_task.__name__
         )
 
     return router

packages/service-library/tests/long_running_tasks/conftest.py

@@ -13,7 +13,7 @@
     TasksManager,
 )
 from settings_library.redis import RedisSettings
-from utils import TEST_CHECK_STALE_INTERVAL_S
+from utils import TEST_CHECK_STALE_INTERVAL_S, NoWebAppLongRunningManager
 
 _logger = logging.getLogger(__name__)
 
@@ -54,3 +54,20 @@ async def _(
     for manager in managers:
         with log_catch(_logger, reraise=False):
             await manager.teardown()
+
+
+@pytest.fixture
+def get_long_running_manager(
+    get_tasks_manager: Callable[
+        [RedisSettings, RedisNamespace | None], Awaitable[TasksManager]
+    ],
+) -> Callable[
+    [RedisSettings, RedisNamespace | None], Awaitable[NoWebAppLongRunningManager]
+]:
+    async def _(
+        redis_settings: RedisSettings, namespace: RedisNamespace | None
+    ) -> NoWebAppLongRunningManager:
+        tasks_manager = await get_tasks_manager(redis_settings, namespace)
+        return NoWebAppLongRunningManager(tasks_manager)
+
+    return _