updating interface

Author: Andrei Neagu

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

@@ -2,7 +2,8 @@
 import functools
 import logging
 import warnings
-from typing import Any, Awaitable, Callable, Final
+from collections.abc import Awaitable, Callable
+from typing import Any, Final
 
 from fastapi import FastAPI, status
 from httpx import AsyncClient, HTTPError
@@ -13,13 +14,8 @@
 from tenacity.stop import stop_after_attempt
 from tenacity.wait import wait_exponential
 
-from ...long_running_tasks._errors import GenericClientError, TaskClientResultError
-from ...long_running_tasks._models import (
-    ClientConfiguration,
-    TaskId,
-    TaskResult,
-    TaskStatus,
-)
+from ...long_running_tasks._errors import GenericClientError
+from ...long_running_tasks._models import ClientConfiguration, TaskId, TaskStatus
 
 DEFAULT_HTTP_REQUESTS_TIMEOUT: Final[PositiveFloat] = 15
 
@@ -85,7 +81,7 @@ def log_it(retry_state: RetryCallState) -> None:
 
 
 def retry_on_http_errors(
-    request_func: Callable[..., Awaitable[Any]]
+    request_func: Callable[..., Awaitable[Any]],
 ) -> Callable[..., Awaitable[Any]]:
     """
     Will retry the request on `httpx.HTTPError`.
@@ -173,10 +169,7 @@ async def get_task_result(
                 body=result.text,
             )
 
-        task_result = TaskResult.model_validate(result.json())
-        if task_result.error is not None:
-            raise TaskClientResultError(message=task_result.error)
-        return task_result.result
+        return result.json()
 
     @retry_on_http_errors
     async def cancel_and_delete_task(

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

@@ -1,6 +1,6 @@
 from typing import Annotated, Any
 
-from fastapi import APIRouter, Depends, Query, Request, status
+from fastapi import APIRouter, Depends, Request, status
 
 from ...long_running_tasks._errors import TaskNotCompletedError, TaskNotFoundError
 from ...long_running_tasks._models import TaskGet, TaskId, TaskResult, TaskStatus
@@ -60,16 +60,10 @@ async def get_task_result(
     request: Request,
     task_id: TaskId,
     tasks_manager: Annotated[TasksManager, Depends(get_tasks_manager)],
-    *,
-    return_exception: Annotated[bool, Query()] = False,
 ) -> TaskResult | Any:
     assert request  # nosec
-    # TODO: refactor this to use same as in https://github.com/ITISFoundation/osparc-simcore/issues/3265
     try:
-        if return_exception:
-            task_result = tasks_manager.get_task_result(task_id, with_task_context=None)
-        else:
-            task_result = tasks_manager.get_task_result_old(task_id=task_id)
+        task_result = tasks_manager.get_task_result(task_id, with_task_context=None)
         await tasks_manager.remove_task(
             task_id, with_task_context=None, reraise_errors=False
         )

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

@@ -97,7 +97,7 @@ async def _wait_for_completion(
 
 @retry(**_DEFAULT_FASTAPI_RETRY_POLICY)
 async def _task_result(session: httpx.AsyncClient, result_url: URL) -> Any:
-    response = await session.get(f"{result_url}", params={"return_exception": True})
+    response = await session.get(f"{result_url}")
     response.raise_for_status()
     if response.status_code != status.HTTP_204_NO_CONTENT:
         return unwrap_envelope_if_required(response.json())

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

@@ -22,7 +22,7 @@
     TaskNotCompletedError,
     TaskNotFoundError,
 )
-from ._models import TaskId, TaskName, TaskResult, TaskStatus, TrackedTask
+from ._models import TaskId, TaskName, TaskStatus, TrackedTask
 
 logger = logging.getLogger(__name__)
 
@@ -241,36 +241,6 @@ def get_task_result(
             # the task was cancelled
             raise TaskCancelledError(task_id=task_id) from exc
 
-    def get_task_result_old(self, task_id: TaskId) -> TaskResult:
-        """
-        returns: the result of the task
-
-        raises TaskNotFoundError if the task cannot be found
-        """
-        tracked_task = self._get_tracked_task(task_id, {})
-
-        if not tracked_task.task.done():
-            raise TaskNotCompletedError(task_id=task_id)
-
-        error: TaskExceptionError | TaskCancelledError
-        try:
-            exception = tracked_task.task.exception()
-            if exception is not None:
-                formatted_traceback = "\n".join(
-                    traceback.format_tb(exception.__traceback__)
-                )
-                error = TaskExceptionError(
-                    task_id=task_id, exception=exception, traceback=formatted_traceback
-                )
-                logger.warning("Task %s finished with error: %s", task_id, f"{error}")
-                return TaskResult(result=None, error=f"{error}")
-        except asyncio.CancelledError:
-            error = TaskCancelledError(task_id=task_id)
-            logger.warning("Task %s was cancelled", task_id)
-            return TaskResult(result=None, error=f"{error}")
-
-        return TaskResult(result=tracked_task.task.result(), error=None)
-
     async def cancel_task(
         self, task_id: TaskId, with_task_context: TaskContext | None
     ) -> None:
@@ -292,7 +262,7 @@ async def _cancel_asyncio_task(
                     await asyncio.wait_for(
                         _await_task(task), timeout=self._cancel_task_timeout_s
                     )
-                except asyncio.TimeoutError:
+                except TimeoutError:
                     logger.warning(
                         "Timed out while awaiting for cancellation of '%s'", reference
                     )
@@ -354,12 +324,12 @@ async def close(self) -> None:
 
 
 class TaskProtocol(Protocol):
-    async def __call__(self, progress: TaskProgress, *args: Any, **kwargs: Any) -> Any:
-        ...
+    async def __call__(
+        self, progress: TaskProgress, *args: Any, **kwargs: Any
+    ) -> Any: ...
 
     @property
-    def __name__(self) -> str:
-        ...
+    def __name__(self) -> str: ...
 
 
 def start_task(

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

@@ -103,9 +103,9 @@ async def _caller(app: FastAPI, client: AsyncClient, **query_kwargs) -> TaskId:
 
 
 @pytest.fixture
-def wait_for_task() -> Callable[
-    [FastAPI, AsyncClient, TaskId, TaskContext], Awaitable[None]
-]:
+def wait_for_task() -> (
+    Callable[[FastAPI, AsyncClient, TaskId, TaskContext], Awaitable[None]]
+):
     async def _waiter(
         app: FastAPI,
         client: AsyncClient,
@@ -183,9 +183,7 @@ async def test_workflow(
     result = await client.get(f"{result_url}")
     # NOTE: this is DIFFERENT than with aiohttp where we return the real result
     assert result.status_code == status.HTTP_200_OK
-    task_result = long_running_tasks.server.TaskResult.model_validate(result.json())
-    assert not task_result.error
-    assert task_result.result == [f"{x}" for x in range(10)]
+    assert result.json() == [f"{x}" for x in range(10)]
     # getting the result again should raise a 404
     result = await client.get(result_url)
     assert result.status_code == status.HTTP_404_NOT_FOUND
@@ -220,19 +218,9 @@ async def test_failing_task_returns_error(
     await wait_for_task(app, client, task_id, {})
     # get the result
     result_url = app.url_path_for("get_task_result", task_id=task_id)
-    result = await client.get(f"{result_url}")
-    assert result.status_code == status.HTTP_200_OK
-    task_result = long_running_tasks.server.TaskResult.model_validate(result.json())
-
-    assert not task_result.result
-    assert task_result.error
-    assert task_result.error.startswith(f"Task {task_id} finished with exception: ")
-    assert 'raise RuntimeError("We were asked to fail!!")' in task_result.error
-    # NOTE: this is not yet happening with fastapi version of long running task
-    # assert "errors" in task_result.error
-    # assert len(task_result.error["errors"]) == 1
-    # assert task_result.error["errors"][0]["code"] == "RuntimeError"
-    # assert task_result.error["errors"][0]["message"] == "We were asked to fail!!"
+    with pytest.raises(RuntimeError) as exec_info:
+        await client.get(f"{result_url}")
+    assert f"{exec_info.value}" == "We were asked to fail!!"
 
 
 async def test_get_results_before_tasks_finishes_returns_404(

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

@@ -2,7 +2,8 @@
 # pylint: disable=unused-argument
 
 import asyncio
-from typing import AsyncIterable, Final
+from collections.abc import AsyncIterable
+from typing import Final
 
 import pytest
 from asgi_lifespan import LifespanManager
@@ -24,9 +25,10 @@
     get_tasks_manager,
 )
 from servicelib.fastapi.long_running_tasks.server import setup as setup_server
-from servicelib.fastapi.long_running_tasks.server import start_task
+from servicelib.fastapi.long_running_tasks.server import (
+    start_task,
+)
 from servicelib.long_running_tasks._errors import (
-    TaskClientResultError,
     TaskClientTimeoutError,
 )
 
@@ -149,16 +151,15 @@ async def test_task_result_task_result_is_an_error(
 
     url = TypeAdapter(AnyHttpUrl).validate_python("http://backgroud.testserver.io/")
     client = Client(app=bg_task_app, async_client=async_client, base_url=url)
-    with pytest.raises(TaskClientResultError) as exec_info:
+    with pytest.raises(RuntimeError) as exec_info:
         async with periodic_task_result(
             client,
             task_id,
             task_timeout=10,
             status_poll_interval=TASK_SLEEP_INTERVAL / 3,
         ):
             pass
-    assert f"{exec_info.value}".startswith(f"Task {task_id} finished with exception:")
-    assert "I am failing as requested" in f"{exec_info.value}"
+    assert f"{exec_info.value}" == "I am failing as requested"
     await _assert_task_removed(async_client, task_id, router_prefix)
 
 

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

@@ -21,7 +21,6 @@
 from servicelib.long_running_tasks._models import (
     ProgressPercent,
     TaskProgress,
-    TaskResult,
     TaskStatus,
 )
 from servicelib.long_running_tasks._task import TasksManager, start_task
@@ -108,8 +107,6 @@ async def test_task_is_auto_removed(
         tasks_manager.get_task_status(task_id, with_task_context=None)
     with pytest.raises(TaskNotFoundError):
         tasks_manager.get_task_result(task_id, with_task_context=None)
-    with pytest.raises(TaskNotFoundError):
-        tasks_manager.get_task_result_old(task_id)
 
 
 async def test_checked_task_is_not_auto_removed(tasks_manager: TasksManager):
@@ -156,9 +153,6 @@ async def test_get_result_of_unfinished_task_raises(tasks_manager: TasksManager)
     with pytest.raises(TaskNotCompletedError):
         tasks_manager.get_task_result(task_id, with_task_context=None)
 
-    with pytest.raises(TaskNotCompletedError):
-        tasks_manager.get_task_result_old(task_id)
-
 
 async def test_unique_task_already_running(tasks_manager: TasksManager):
     async def unique_task(task_progress: TaskProgress):
@@ -214,13 +208,6 @@ async def test_get_result(tasks_manager: TasksManager):
     assert result == 42
 
 
-async def test_get_result_old(tasks_manager: TasksManager):
-    task_id = start_task(tasks_manager=tasks_manager, task=fast_background_task)
-    await asyncio.sleep(0.1)
-    result = tasks_manager.get_task_result_old(task_id)
-    assert result == TaskResult(result=42, error=None)
-
-
 async def test_get_result_missing(tasks_manager: TasksManager):
     with pytest.raises(TaskNotFoundError) as exec_info:
         tasks_manager.get_task_result("missing_task_id", with_task_context=None)
@@ -238,20 +225,6 @@ async def test_get_result_finished_with_error(tasks_manager: TasksManager):
         tasks_manager.get_task_result(task_id, with_task_context=None)
 
 
-async def test_get_result_old_finished_with_error(tasks_manager: TasksManager):
-    task_id = start_task(tasks_manager=tasks_manager, task=failing_background_task)
-    # wait for result
-    async for attempt in AsyncRetrying(**_RETRY_PARAMS):
-        with attempt:
-            assert tasks_manager.get_task_status(task_id, with_task_context=None).done
-
-    task_result = tasks_manager.get_task_result_old(task_id)
-    assert task_result.result is None
-    assert task_result.error is not None
-    assert task_result.error.startswith(f"Task {task_id} finished with exception:")
-    assert "failing asap" in task_result.error
-
-
 async def test_get_result_task_was_cancelled_multiple_times(
     tasks_manager: TasksManager,
 ):
@@ -270,23 +243,6 @@ async def test_get_result_task_was_cancelled_multiple_times(
         tasks_manager.get_task_result(task_id, with_task_context=None)
 
 
-async def test_get_result_old_task_was_cancelled_multiple_times(
-    tasks_manager: TasksManager,
-):
-    task_id = start_task(
-        tasks_manager=tasks_manager,
-        task=a_background_task,
-        raise_when_finished=False,
-        total_sleep=10,
-    )
-    for _ in range(5):
-        await tasks_manager.cancel_task(task_id, with_task_context=None)
-
-    task_result = tasks_manager.get_task_result_old(task_id)
-    assert task_result.result is None
-    assert task_result.error == f"Task {task_id} was cancelled before completing"
-
-
 async def test_remove_task(tasks_manager: TasksManager):
     task_id = start_task(
         tasks_manager=tasks_manager,
@@ -300,8 +256,6 @@ async def test_remove_task(tasks_manager: TasksManager):
         tasks_manager.get_task_status(task_id, with_task_context=None)
     with pytest.raises(TaskNotFoundError):
         tasks_manager.get_task_result(task_id, with_task_context=None)
-    with pytest.raises(TaskNotFoundError):
-        tasks_manager.get_task_result_old(task_id)
 
 
 async def test_remove_task_with_task_context(tasks_manager: TasksManager):