wired task_start via common interface

Author: Andrei Neagu

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

@@ -9,6 +9,7 @@
 from pydantic import AnyHttpUrl, TypeAdapter
 
 from ...aiohttp import status
+from ...long_running_tasks import http_endpoint_responses
 from ...long_running_tasks.constants import (
     DEFAULT_STALE_TASK_CHECK_INTERVAL,
     DEFAULT_STALE_TASK_DETECT_TIMEOUT,
@@ -55,7 +56,8 @@ async def start_long_running_task(
     task_name = _create_task_name_from_request(request_)
     task_id = None
     try:
-        task_id = long_running_manager.tasks_manager.start_task(
+        task_id = await http_endpoint_responses.start_task(
+            long_running_manager.tasks_manager,
             registerd_task_name,
             fire_and_forget=fire_and_forget,
             task_context=task_context,

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

@@ -1,7 +1,7 @@
 from typing import Any
 
 from .models import TaskBase, TaskId, TaskStatus
-from .task import TaskContext, TasksManager, TrackedTask
+from .task import RegisteredTaskName, TaskContext, TasksManager, TrackedTask
 
 
 def list_tasks(
@@ -16,6 +16,7 @@ def list_tasks(
 def get_task_status(
     tasks_manager: TasksManager, task_context: TaskContext | None, task_id: TaskId
 ) -> TaskStatus:
+    """returns the status of a task"""
     return tasks_manager.get_task_status(
         task_id=task_id, with_task_context=task_context
     )
@@ -24,6 +25,7 @@ def get_task_status(
 async def get_task_result(
     tasks_manager: TasksManager, task_context: TaskContext | None, task_id: TaskId
 ) -> Any:
+    """retruns the result of a task, which is directly whatever the remove hanlder returned"""
     try:
         return tasks_manager.get_task_result(
             task_id=task_id, with_task_context=task_context
@@ -38,23 +40,50 @@ async def get_task_result(
 async def remove_task(
     tasks_manager: TasksManager, task_context: TaskContext | None, task_id: TaskId
 ) -> None:
+    """removes / cancels a task"""
     await tasks_manager.remove_task(task_id, with_task_context=task_context)
 
 
-# TODO reroute start via this with registration of hanlders
+async def start_task(
+    tasks_manager: TasksManager,
+    registered_task_name: RegisteredTaskName,
+    *,
+    unique: bool = False,
+    task_context: TaskContext | None = None,
+    task_name: str | None = None,
+    fire_and_forget: bool = False,
+    **task_kwargs: Any,
+) -> TaskId:
+    """
+    Creates a background task from an async function.
 
-# TODO: to support this we need handler registration capability in order to figure out if they can be started
-# this should be done via the final TasksManager
-# - also
+    An asyncio task will be created out of it by injecting a `TaskProgress` as the first
+    positional argument and adding all `handler_kwargs` as named parameters.
 
-# TODO: this should be used to start everywhere not the client's start_task actually,
-# this way we completly isolate everything and are allowed to call it form everywhere
-# async def start_task(
-#     tasks_manager: TasksManager,
-#     task: TaskContext | None,
-#     task_context: TaskContext | None = None,
-#     **task_kwargs: Any,
-# ) -> TaskId:
-#     return tasks_manager.start_task(
-#         task=task, task_context=task_context, **task_kwargs
-#     )
+    NOTE: the progress is automatically bounded between 0 and 1
+    NOTE: the `task` name must be unique in the module, otherwise when using
+        the unique parameter is True, it will not be able to distinguish between
+        them.
+
+    Args:
+        tasks_manager (TasksManager): the tasks manager
+        task (TaskProtocol): the tasks to be run in the background
+        unique (bool, optional): If True, then only one such named task may be run. Defaults to False.
+        task_context (Optional[TaskContext], optional): a task context storage can be retrieved during the task lifetime. Defaults to None.
+        task_name (Optional[str], optional): optional task name. Defaults to None.
+        fire_and_forget: if True, then the task will not be cancelled if the status is never called
+
+    Raises:
+        TaskAlreadyRunningError: if unique is True, will raise if more than 1 such named task is started
+
+    Returns:
+        TaskId: the task unique identifier
+    """
+    return tasks_manager.start_task(
+        registered_task_name,
+        unique=unique,
+        task_context=task_context,
+        task_name=task_name,
+        fire_and_forget=fire_and_forget,
+        **task_kwargs,
+    )

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

@@ -335,37 +335,12 @@ def start_task(
         self,
         registered_task_name: RegisteredTaskName,
         *,
-        unique: bool = False,
-        task_context: TaskContext | None = None,
-        task_name: str | None = None,
-        fire_and_forget: bool = False,
+        unique: bool,
+        task_context: TaskContext | None,
+        task_name: str | None,
+        fire_and_forget: bool,
         **task_kwargs: Any,
     ) -> TaskId:
-        """
-        Creates a background task from an async function.
-
-        An asyncio task will be created out of it by injecting a `TaskProgress` as the first
-        positional argument and adding all `handler_kwargs` as named parameters.
-
-        NOTE: the progress is automatically bounded between 0 and 1
-        NOTE: the `task` name must be unique in the module, otherwise when using
-            the unique parameter is True, it will not be able to distinguish between
-            them.
-
-        Args:
-            tasks_manager (TasksManager): the tasks manager
-            task (TaskProtocol): the tasks to be run in the background
-            unique (bool, optional): If True, then only one such named task may be run. Defaults to False.
-            task_context (Optional[TaskContext], optional): a task context storage can be retrieved during the task lifetime. Defaults to None.
-            task_name (Optional[str], optional): optional task name. Defaults to None.
-            fire_and_forget: if True, then the task will not be cancelled if the status is never called
-
-        Raises:
-            TaskAlreadyRunningError: if unique is True, will raise if more than 1 such named task is started
-
-        Returns:
-            TaskId: the task unique identifier
-        """
         if registered_task_name not in TaskRegistry.REGISTERED_TASKS:
             raise TaskNotRegisteredError(task_name=registered_task_name)
 

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

@@ -25,6 +25,7 @@
     get_long_running_manager,
 )
 from servicelib.fastapi.long_running_tasks.server import setup as setup_server
+from servicelib.long_running_tasks import http_endpoint_responses
 from servicelib.long_running_tasks.models import (
     TaskGet,
     TaskId,
@@ -78,7 +79,8 @@ async def create_string_list_task(
         *,
         fail: bool = False,
     ) -> TaskId:
-        return long_running_manager.tasks_manager.start_task(
+        return await http_endpoint_responses.start_task(
+            long_running_manager.tasks_manager,
             _string_list_task.__name__,
             num_strings=num_strings,
             sleep_time=sleep_time,

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

@@ -16,6 +16,7 @@
 from servicelib.fastapi.long_running_tasks.client import setup as setup_client
 from servicelib.fastapi.long_running_tasks.server import get_long_running_manager
 from servicelib.fastapi.long_running_tasks.server import setup as setup_server
+from servicelib.long_running_tasks import http_endpoint_responses
 from servicelib.long_running_tasks.errors import (
     TaskClientTimeoutError,
 )
@@ -68,16 +69,18 @@ async def create_task_user_defined_route(
             FastAPILongRunningManager, Depends(get_long_running_manager)
         ],
     ) -> TaskId:
-        return long_running_manager.tasks_manager.start_task(a_test_task.__name__)
+        return await http_endpoint_responses.start_task(
+            long_running_manager.tasks_manager, a_test_task.__name__
+        )
 
     @router.get("/api/failing", status_code=status.HTTP_200_OK)
     async def create_task_which_fails(
         long_running_manager: Annotated[
             FastAPILongRunningManager, Depends(get_long_running_manager)
         ],
     ) -> TaskId:
-        return long_running_manager.tasks_manager.start_task(
-            a_failing_test_task.__name__
+        return await http_endpoint_responses.start_task(
+            long_running_manager.tasks_manager, a_failing_test_task.__name__
         )
 
     return router

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

@@ -12,6 +12,7 @@
 
 import pytest
 from faker import Faker
+from servicelib.long_running_tasks import http_endpoint_responses
 from servicelib.long_running_tasks.errors import (
     TaskAlreadyRunningError,
     TaskCancelledError,
@@ -84,7 +85,8 @@ async def tasks_manager() -> AsyncIterator[TasksManager]:
 async def test_task_is_auto_removed(
     tasks_manager: TasksManager, check_task_presence_before: bool
 ):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10 * TEST_CHECK_STALE_INTERVAL_S,
@@ -110,7 +112,8 @@ async def test_task_is_auto_removed(
 
 
 async def test_checked_task_is_not_auto_removed(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=5 * TEST_CHECK_STALE_INTERVAL_S,
@@ -124,7 +127,8 @@ async def test_checked_task_is_not_auto_removed(tasks_manager: TasksManager):
 
 
 async def test_fire_and_forget_task_is_not_auto_removed(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=5 * TEST_CHECK_STALE_INTERVAL_S,
@@ -142,7 +146,8 @@ async def test_fire_and_forget_task_is_not_auto_removed(tasks_manager: TasksMana
 
 
 async def test_get_result_of_unfinished_task_raises(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=5 * TEST_CHECK_STALE_INTERVAL_S,
@@ -158,11 +163,15 @@ async def unique_task(progress: TaskProgress):
 
     TaskRegistry.register(unique_task)
 
-    tasks_manager.start_task(unique_task.__name__, unique=True)
+    await http_endpoint_responses.start_task(
+        tasks_manager, unique_task.__name__, unique=True
+    )
 
     # ensure unique running task regardless of how many times it gets started
     with pytest.raises(TaskAlreadyRunningError) as exec_info:
-        tasks_manager.start_task(unique_task.__name__, unique=True)
+        await http_endpoint_responses.start_task(
+            tasks_manager, unique_task.__name__, unique=True
+        )
     assert "must be unique, found: " in f"{exec_info.value}"
 
     TaskRegistry.unregister(unique_task)
@@ -175,7 +184,9 @@ async def not_unique_task(progress: TaskProgress):
     TaskRegistry.register(not_unique_task)
 
     for _ in range(5):
-        tasks_manager.start_task(not_unique_task.__name__)
+        await http_endpoint_responses.start_task(
+            tasks_manager, not_unique_task.__name__
+        )
 
     TaskRegistry.unregister(not_unique_task)
 
@@ -188,7 +199,8 @@ def test_get_task_id(tasks_manager: TasksManager, faker: Faker, is_unique: bool)
 
 
 async def test_get_status(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -208,7 +220,9 @@ async def test_get_status_missing(tasks_manager: TasksManager):
 
 
 async def test_get_result(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(fast_background_task.__name__)
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager, fast_background_task.__name__
+    )
     await asyncio.sleep(0.1)
     result = tasks_manager.get_task_result(task_id, with_task_context=None)
     assert result == 42
@@ -221,7 +235,9 @@ async def test_get_result_missing(tasks_manager: TasksManager):
 
 
 async def test_get_result_finished_with_error(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(failing_background_task.__name__)
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager, failing_background_task.__name__
+    )
     # wait for result
     async for attempt in AsyncRetrying(**_RETRY_PARAMS):
         with attempt:
@@ -234,7 +250,8 @@ async def test_get_result_finished_with_error(tasks_manager: TasksManager):
 async def test_get_result_task_was_cancelled_multiple_times(
     tasks_manager: TasksManager,
 ):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -249,7 +266,8 @@ async def test_get_result_task_was_cancelled_multiple_times(
 
 
 async def test_remove_task(tasks_manager: TasksManager):
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -264,7 +282,8 @@ async def test_remove_task(tasks_manager: TasksManager):
 
 async def test_remove_task_with_task_context(tasks_manager: TasksManager):
     TASK_CONTEXT = {"some_context": "some_value"}
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -296,7 +315,8 @@ async def test_remove_unknown_task(tasks_manager: TasksManager):
 
 async def test_cancel_task_with_task_context(tasks_manager: TasksManager):
     TASK_CONTEXT = {"some_context": "some_value"}
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -322,7 +342,8 @@ async def test_list_tasks(tasks_manager: TasksManager):
     task_ids = []
     for _ in range(NUM_TASKS):
         task_ids.append(  # noqa: PERF401
-            tasks_manager.start_task(
+            await http_endpoint_responses.start_task(
+                tasks_manager,
                 a_background_task.__name__,
                 raise_when_finished=False,
                 total_sleep=10,
@@ -337,18 +358,21 @@ async def test_list_tasks(tasks_manager: TasksManager):
 
 
 async def test_list_tasks_filtering(tasks_manager: TasksManager):
-    tasks_manager.start_task(
+    await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
     )
-    tasks_manager.start_task(
+    await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
         task_context={"user_id": 213},
     )
-    tasks_manager.start_task(
+    await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -376,7 +400,8 @@ async def test_list_tasks_filtering(tasks_manager: TasksManager):
 
 async def test_define_task_name(tasks_manager: TasksManager, faker: Faker):
     task_name = faker.name()
-    task_id = tasks_manager.start_task(
+    task_id = await http_endpoint_responses.start_task(
+        tasks_manager,
         a_background_task.__name__,
         raise_when_finished=False,
         total_sleep=10,
@@ -387,4 +412,4 @@ async def test_define_task_name(tasks_manager: TasksManager, faker: Faker):
 
 async def test_start_not_registered_task(tasks_manager: TasksManager):
     with pytest.raises(TaskNotRegisteredError):
-        tasks_manager.start_task("not_registered_task")
+        await http_endpoint_responses.start_task(tasks_manager, "not_registered_task")