de-duplicate

Author: sanderegg

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

@@ -1,8 +1,7 @@
 import asyncio
 import logging
-from collections.abc import AsyncGenerator, Coroutine
-from dataclasses import dataclass
-from typing import Any, Final, TypeAlias
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from aiohttp import ClientConnectionError, ClientSession
 from tenacity import TryAgain, retry
@@ -13,17 +12,15 @@
 from tenacity.wait import wait_random_exponential
 from yarl import URL
 
+from ...long_running_tasks._constants import DEFAULT_POLL_INTERVAL_S, HOUR
+from ...long_running_tasks._models import LRTask, RequestBody
 from ...rest_responses import unwrap_envelope_if_required
 from .. import status
 from .server import TaskGet, TaskId, TaskProgress, TaskStatus
 
 _logger = logging.getLogger(__name__)
 
-RequestBody: TypeAlias = Any
 
-_MINUTE: Final[int] = 60  # in secs
-_HOUR: Final[int] = 60 * _MINUTE  # in secs
-_DEFAULT_POLL_INTERVAL_S: Final[float] = 1
 _DEFAULT_AIOHTTP_RETRY_POLICY: dict[str, Any] = {
     "retry": retry_if_exception_type(ClientConnectionError),
     "wait": wait_random_exponential(max=20),
@@ -64,9 +61,7 @@ async def _wait_for_completion(
                 if not task_status.done:
                     await asyncio.sleep(
                         float(
-                            response.headers.get(
-                                "retry-after", _DEFAULT_POLL_INTERVAL_S
-                            )
+                            response.headers.get("retry-after", DEFAULT_POLL_INTERVAL_S)
                         )
                     )
                     msg = f"{task_id=}, {task_status.started=} has status: '{task_status.task_progress.message}' {task_status.task_progress.percent}%"
@@ -93,26 +88,11 @@ async def _abort_task(session: ClientSession, abort_url: URL) -> None:
         response.raise_for_status()
 
 
-@dataclass(frozen=True)
-class LRTask:
-    progress: TaskProgress
-    _result: Coroutine[Any, Any, Any] | None = None
-
-    def done(self) -> bool:
-        return self._result is not None
-
-    async def result(self) -> Any:
-        if not self._result:
-            msg = "No result ready!"
-            raise ValueError(msg)
-        return await self._result
-
-
 async def long_running_task_request(
     session: ClientSession,
     url: URL,
     json: RequestBody | None = None,
-    client_timeout: int = 1 * _HOUR,
+    client_timeout: int = 1 * HOUR,
 ) -> AsyncGenerator[LRTask, None]:
     """Will use the passed `ClientSession` to call an oSparc long
     running task `url` passing `json` as request body.

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

@@ -6,7 +6,7 @@
 import logging
 from collections.abc import AsyncGenerator, Coroutine
 from dataclasses import dataclass
-from typing import Any, Final, TypeAlias
+from typing import Any
 
 import httpx
 from fastapi import status
@@ -23,12 +23,15 @@
 )
 from yarl import URL
 
+from ...long_running_tasks._constants import DEFAULT_POLL_INTERVAL_S, HOUR
 from ...long_running_tasks._errors import TaskClientResultError
 from ...long_running_tasks._models import (
     ClientConfiguration,
+    LRTask,
     ProgressCallback,
     ProgressMessage,
     ProgressPercent,
+    RequestBody,
 )
 from ...long_running_tasks._task import TaskId, TaskResult
 from ...rest_responses import unwrap_envelope_if_required
@@ -37,11 +40,7 @@
 
 _logger = logging.getLogger(__name__)
 
-RequestBody: TypeAlias = Any
 
-_MINUTE: Final[int] = 60  # in secs
-_HOUR: Final[int] = 60 * _MINUTE  # in secs
-_DEFAULT_POLL_INTERVAL_S: Final[float] = 1
 _DEFAULT_FASTAPI_RETRY_POLICY: dict[str, Any] = {
     "retry": retry_if_exception_type(httpx.RequestError),
     "wait": wait_random_exponential(max=20),
@@ -85,9 +84,7 @@ async def _wait_for_completion(
                 if not task_status.done:
                     await asyncio.sleep(
                         float(
-                            response.headers.get(
-                                "retry-after", _DEFAULT_POLL_INTERVAL_S
-                            )
+                            response.headers.get("retry-after", DEFAULT_POLL_INTERVAL_S)
                         )
                     )
                     msg = f"{task_id=}, {task_status.started=} has status: '{task_status.task_progress.message}' {task_status.task_progress.percent}%"
@@ -114,26 +111,11 @@ async def _abort_task(session: httpx.AsyncClient, abort_url: URL) -> None:
     response.raise_for_status()
 
 
-@dataclass(frozen=True)
-class LRTask:
-    progress: TaskProgress
-    _result: Coroutine[Any, Any, Any] | None = None
-
-    def done(self) -> bool:
-        return self._result is not None
-
-    async def result(self) -> Any:
-        if not self._result:
-            msg = "No result ready!"
-            raise ValueError(msg)
-        return await self._result
-
-
 async def long_running_task_request(
     session: httpx.AsyncClient,
     url: URL,
     json: RequestBody | None = None,
-    client_timeout: int = 1 * _HOUR,
+    client_timeout: int = 1 * HOUR,
 ) -> AsyncGenerator[LRTask, None]:
     """Will use the passed `httpx.AsyncClient` to call an oSparc long
     running task `url` passing `json` as request body.
@@ -167,16 +149,17 @@ async def long_running_task_request(
 
 
 __all__: tuple[str, ...] = (
-    "DEFAULT_HTTP_REQUESTS_TIMEOUT",
     "Client",
     "ClientConfiguration",
+    "DEFAULT_HTTP_REQUESTS_TIMEOUT",
+    "LRTask",
+    "periodic_task_result",
     "ProgressCallback",
     "ProgressMessage",
     "ProgressPercent",
+    "setup",
     "TaskClientResultError",
     "TaskId",
     "TaskResult",
-    "periodic_task_result",
-    "setup",
 )
 # nopycln: file

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

@@ -0,0 +1,5 @@
+from typing import Final
+
+MINUTE: Final[int] = 60  # in secs
+HOUR: Final[int] = 60 * MINUTE  # in secs
+DEFAULT_POLL_INTERVAL_S: Final[float] = 1

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

@@ -1,6 +1,7 @@
 # mypy: disable-error-code=truthy-function
 from asyncio import Task
 from collections.abc import Awaitable, Callable, Coroutine
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, TypeAlias
 
@@ -25,6 +26,8 @@
     [ProgressMessage, ProgressPercent | None, TaskId], Awaitable[None]
 ]
 
+RequestBody: TypeAlias = Any
+
 
 class TrackedTask(BaseModel):
     task_id: str
@@ -56,18 +59,33 @@ class ClientConfiguration(BaseModel):
     default_timeout: PositiveFloat
 
 
+@dataclass(frozen=True)
+class LRTask:
+    progress: TaskProgress
+    _result: Coroutine[Any, Any, Any] | None = None
+
+    def done(self) -> bool:
+        return self._result is not None
+
+    async def result(self) -> Any:
+        if not self._result:
+            msg = "No result ready!"
+            raise ValueError(msg)
+        return await self._result
+
+
 # explicit export of models for api-schemas
 
 assert TaskResult  # nosec
 assert TaskGet  # nosec
 assert TaskStatus  # nosec
 
 __all__: tuple[str, ...] = (
+    "ProgressMessage",
+    "ProgressPercent",
     "TaskGet",
     "TaskId",
+    "TaskProgress",
     "TaskResult",
     "TaskStatus",
-    "TaskProgress",
-    "ProgressPercent",
-    "ProgressMessage",
 )

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

@@ -2,7 +2,7 @@
 # pylint: disable=unused-argument
 
 import asyncio
-from typing import Callable
+from collections.abc import Callable
 
 import pytest
 from aiohttp import ClientResponseError, web
@@ -36,7 +36,6 @@ def client(
     unused_tcp_port_factory: Callable,
     app: web.Application,
 ) -> TestClient:
-
     return event_loop.run_until_complete(
         aiohttp_client(app, server_kwargs={"port": unused_tcp_port_factory()})
     )