♻️ preparing TasksManager's interface to be extracted into a common interface

packages/models-library/src/models_library/api_schemas_long_running_tasks/tasks.py

@@ -1,8 +1,7 @@
-import urllib.parse
 from datetime import datetime
 from typing import Any
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel
 
 from .base import TaskId, TaskProgress
 
@@ -20,15 +19,9 @@ class TaskResult(BaseModel):
 
 class TaskBase(BaseModel):
     task_id: TaskId
-    task_name: str
 
 
 class TaskGet(TaskBase):
     status_href: str
     result_href: str
     abort_href: str
-
-    @field_validator("task_name")
-    @classmethod
-    def unquote_str(cls, v) -> str:
-        return urllib.parse.unquote(v)

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

@@ -4,7 +4,7 @@
 from pydantic import BaseModel
 from servicelib.aiohttp import status
 
-from ...long_running_tasks import http_endpoint_responses
+from ...long_running_tasks import lrt_api
 from ...long_running_tasks.models import TaskGet, TaskId, TaskStatus
 from ..requests_validation import parse_request_path_parameters_as
 from ..rest_responses import create_data_response
@@ -24,12 +24,11 @@ async def list_tasks(request: web.Request) -> web.Response:
         [
             TaskGet(
                 task_id=t.task_id,
-                task_name=t.task_name,
                 status_href=f"{request.app.router['get_task_status'].url_for(task_id=t.task_id)}",
                 result_href=f"{request.app.router['get_task_result'].url_for(task_id=t.task_id)}",
                 abort_href=f"{request.app.router['cancel_and_delete_task'].url_for(task_id=t.task_id)}",
             )
-            for t in http_endpoint_responses.list_tasks(
+            for t in lrt_api.list_tasks(
                 long_running_manager.tasks_manager,
                 long_running_manager.get_task_context(request),
             )
@@ -42,7 +41,7 @@ async def get_task_status(request: web.Request) -> web.Response:
     path_params = parse_request_path_parameters_as(_PathParam, request)
     long_running_manager = get_long_running_manager(request.app)
 
-    task_status: TaskStatus = http_endpoint_responses.get_task_status(
+    task_status: TaskStatus = lrt_api.get_task_status(
         long_running_manager.tasks_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
@@ -56,7 +55,7 @@ async def get_task_result(request: web.Request) -> web.Response | Any:
     long_running_manager = get_long_running_manager(request.app)
 
     # NOTE: this might raise an exception that will be catched by the _error_handlers
-    return await http_endpoint_responses.get_task_result(
+    return await lrt_api.get_task_result(
         long_running_manager.tasks_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,
@@ -68,7 +67,7 @@ async def cancel_and_delete_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 http_endpoint_responses.remove_task(
+    await lrt_api.remove_task(
         long_running_manager.tasks_manager,
         long_running_manager.get_task_context(request),
         path_params.task_id,

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

@@ -9,12 +9,13 @@
 from pydantic import AnyHttpUrl, TypeAdapter
 
 from ...aiohttp import status
+from ...long_running_tasks import lrt_api
 from ...long_running_tasks.constants import (
     DEFAULT_STALE_TASK_CHECK_INTERVAL,
     DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 )
 from ...long_running_tasks.models import TaskGet
-from ...long_running_tasks.task import TaskContext, TaskProtocol, start_task
+from ...long_running_tasks.task import RegisteredTaskName, TaskContext
 from ..typing_extension import Handler
 from . import _routes
 from ._constants import (
@@ -25,11 +26,11 @@
 from ._manager import AiohttpLongRunningManager, get_long_running_manager
 
 
-def no_ops_decorator(handler: Handler):
+def _no_ops_decorator(handler: Handler):
     return handler
 
 
-def no_task_context_decorator(handler: Handler):
+def _no_task_context_decorator(handler: Handler):
     @wraps(handler)
     async def _wrap(request: web.Request):
         request[RQT_LONG_RUNNING_TASKS_CONTEXT_KEY] = {}
@@ -45,7 +46,7 @@ def _create_task_name_from_request(request: web.Request) -> str:
 async def start_long_running_task(
     # NOTE: positional argument are suffixed with "_" to avoid name conflicts with "task_kwargs" keys
     request_: web.Request,
-    task_: TaskProtocol,
+    registerd_task_name: RegisteredTaskName,
     *,
     fire_and_forget: bool = False,
     task_context: TaskContext,
@@ -55,9 +56,9 @@ async def start_long_running_task(
     task_name = _create_task_name_from_request(request_)
     task_id = None
     try:
-        task_id = start_task(
+        task_id = await lrt_api.start_task(
             long_running_manager.tasks_manager,
-            task_,
+            registerd_task_name,
             fire_and_forget=fire_and_forget,
             task_context=task_context,
             task_name=task_name,
@@ -78,7 +79,6 @@ async def start_long_running_task(
         )
         task_get = TaskGet(
             task_id=task_id,
-            task_name=task_name,
             status_href=f"{status_url}",
             result_href=f"{result_url}",
             abort_href=f"{abort_url}",
@@ -121,8 +121,8 @@ def setup(
     app: web.Application,
     *,
     router_prefix: str,
-    handler_check_decorator: Callable = no_ops_decorator,
-    task_request_context_decorator: Callable = no_task_context_decorator,
+    handler_check_decorator: Callable = _no_ops_decorator,
+    task_request_context_decorator: Callable = _no_task_context_decorator,
     stale_task_check_interval: datetime.timedelta = DEFAULT_STALE_TASK_CHECK_INTERVAL,
     stale_task_detect_timeout: datetime.timedelta = DEFAULT_STALE_TASK_DETECT_TIMEOUT,
 ) -> None:

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

@@ -2,7 +2,7 @@
 
 from fastapi import APIRouter, Depends, Request, status
 
-from ...long_running_tasks import http_endpoint_responses
+from ...long_running_tasks import lrt_api
 from ...long_running_tasks.models import TaskGet, TaskId, TaskResult, TaskStatus
 from ..requests_decorators import cancel_on_disconnect
 from ._dependencies import get_long_running_manager
@@ -23,14 +23,13 @@ async def list_tasks(
     return [
         TaskGet(
             task_id=t.task_id,
-            task_name=t.task_name,
             status_href=str(request.url_for("get_task_status", task_id=t.task_id)),
             result_href=str(request.url_for("get_task_result", task_id=t.task_id)),
             abort_href=str(
                 request.url_for("cancel_and_delete_task", task_id=t.task_id)
             ),
         )
-        for t in http_endpoint_responses.list_tasks(
+        for t in lrt_api.list_tasks(
             long_running_manager.tasks_manager, task_context=None
         )
     ]
@@ -52,7 +51,7 @@ async def get_task_status(
     ],
 ) -> TaskStatus:
     assert request  # nosec
-    return http_endpoint_responses.get_task_status(
+    return lrt_api.get_task_status(
         long_running_manager.tasks_manager, task_context=None, task_id=task_id
     )
 
@@ -75,7 +74,7 @@ async def get_task_result(
     ],
 ) -> TaskResult | Any:
     assert request  # nosec
-    return await http_endpoint_responses.get_task_result(
+    return await lrt_api.get_task_result(
         long_running_manager.tasks_manager, task_context=None, task_id=task_id
     )
 
@@ -98,6 +97,6 @@ async def cancel_and_delete_task(
     ],
 ) -> None:
     assert request  # nosec
-    await http_endpoint_responses.remove_task(
+    await lrt_api.remove_task(
         long_running_manager.tasks_manager, task_context=None, task_id=task_id
     )

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

@@ -5,6 +5,13 @@ class BaseLongRunningError(OsparcErrorMixin, Exception):
     """base exception for this module"""
 
 
+class TaskNotRegisteredError(BaseLongRunningError):
+    msg_template: str = (
+        "notask with task_name='{task_name}' was found in the task registry. "
+        "Make sure it's registered before starting it."
+    )
+
+
 class TaskAlreadyRunningError(BaseLongRunningError):
     msg_template: str = "{task_name} must be unique, found: '{managed_task}'"
 

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

@@ -0,0 +1,89 @@
+from typing import Any
+
+from .models import TaskBase, TaskId, TaskStatus
+from .task import RegisteredTaskName, TaskContext, TasksManager, TrackedTask
+
+
+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.
+
+    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
+    """
+    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,
+    )
+
+
+def list_tasks(
+    tasks_manager: TasksManager, task_context: TaskContext | None
+) -> list[TaskBase]:
+    tracked_tasks: list[TrackedTask] = tasks_manager.list_tasks(
+        with_task_context=task_context
+    )
+    return [TaskBase(task_id=t.task_id) for t in tracked_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
+    )
+
+
+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
+        )
+    finally:
+        # the task is always removed even if an error occurs
+        await tasks_manager.remove_task(
+            task_id, with_task_context=task_context, reraise_errors=False
+        )
+
+
+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)

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

@@ -19,8 +19,6 @@
 )
 from pydantic import BaseModel, ConfigDict, Field, PositiveFloat
 
-TaskName: TypeAlias = str
-
 TaskType: TypeAlias = Callable[..., Coroutine[Any, Any, Any]]
 
 ProgressCallback: TypeAlias = Callable[
@@ -33,7 +31,6 @@
 class TrackedTask(BaseModel):
     task_id: str
     task: Task
-    task_name: TaskName
     task_progress: TaskProgress
     # NOTE: this context lifetime is with the tracked task (similar to aiohttp storage concept)
     task_context: dict[str, Any]