Merge branch 'master' into request/hide-osparc-integration

Author: odeimaiz

packages/models-library/src/models_library/progress_bar.py

@@ -2,14 +2,12 @@
 
 from pydantic import BaseModel, ConfigDict
 
-from .basic_types import IDStr
-
 # NOTE: keep a list of possible unit, and please use correct official unit names
 ProgressUnit: TypeAlias = Literal["Byte"]
 
 
 class ProgressStructuredMessage(BaseModel):
-    description: IDStr
+    description: str
     current: float
     total: int
     unit: str | None = None
@@ -51,6 +49,7 @@ class ProgressStructuredMessage(BaseModel):
 class ProgressReport(BaseModel):
     actual_value: float
     total: float = 1.0
+    attempt: int = 0
     unit: ProgressUnit | None = UNITLESS
     message: ProgressStructuredMessage | None = None
 

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

@@ -1,40 +1,40 @@
 import asyncio
-from collections.abc import AsyncGenerator, Coroutine
-from dataclasses import dataclass
-from typing import Any, Final, TypeAlias
+import logging
+from collections.abc import AsyncGenerator
+from typing import Any
 
 from aiohttp import ClientConnectionError, ClientSession
 from tenacity import TryAgain, retry
 from tenacity.asyncio import AsyncRetrying
+from tenacity.before_sleep import before_sleep_log
 from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_random_exponential
 from yarl import URL
 
-from ...rest_responses import unwrap_envelope
+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
 
-RequestBody: TypeAlias = Any
+_logger = logging.getLogger(__name__)
+
 
-_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),
     "stop": stop_after_delay(60),
     "reraise": True,
+    "before_sleep": before_sleep_log(_logger, logging.INFO),
 }
 
 
 @retry(**_DEFAULT_AIOHTTP_RETRY_POLICY)
 async def _start(session: ClientSession, url: URL, json: RequestBody | None) -> TaskGet:
     async with session.post(url, json=json) as response:
         response.raise_for_status()
-        data, error = unwrap_envelope(await response.json())
-    assert not error  # nosec
-    assert data is not None  # nosec
+        data = unwrap_envelope_if_required(await response.json())
     return TaskGet.model_validate(data)
 
 
@@ -50,21 +50,18 @@ async def _wait_for_completion(
             stop=stop_after_delay(client_timeout),
             reraise=True,
             retry=retry_if_exception_type(TryAgain),
+            before_sleep=before_sleep_log(_logger, logging.DEBUG),
         ):
             with attempt:
                 async with session.get(status_url) as response:
                     response.raise_for_status()
-                    data, error = unwrap_envelope(await response.json())
-                    assert not error  # nosec
-                    assert data is not None  # nosec
+                    data = unwrap_envelope_if_required(await response.json())
                 task_status = TaskStatus.model_validate(data)
                 yield task_status.task_progress
                 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}%"
@@ -81,42 +78,21 @@ async def _task_result(session: ClientSession, result_url: URL) -> Any:
     async with session.get(result_url) as response:
         response.raise_for_status()
         if response.status != status.HTTP_204_NO_CONTENT:
-            data, error = unwrap_envelope(await response.json())
-            assert not error  # nosec
-            assert data  # nosec
-            return data
+            return unwrap_envelope_if_required(await response.json())
         return None
 
 
 @retry(**_DEFAULT_AIOHTTP_RETRY_POLICY)
 async def _abort_task(session: ClientSession, abort_url: URL) -> None:
     async with session.delete(abort_url) as response:
         response.raise_for_status()
-        data, error = unwrap_envelope(await response.json())
-        assert not error  # nosec
-        assert not data  # nosec
-
-
-@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.
@@ -147,3 +123,6 @@ async def long_running_task_request(
         if task:
             await _abort_task(session, URL(task.abort_href))
         raise
+
+
+__all__: tuple[str, ...] = ("LRTask",)

packages/service-library/src/servicelib/archiving_utils/_interface_7zip.py

@@ -10,7 +10,6 @@
 from typing import Final
 
 import tqdm
-from models_library.basic_types import IDStr
 from pydantic import NonNegativeInt
 from servicelib.logging_utils import log_catch
 from tqdm.contrib.logging import tqdm_logging_redirect
@@ -199,7 +198,7 @@ async def archive_dir(
 ) -> None:
     if progress_bar is None:
         progress_bar = ProgressBarData(
-            num_steps=1, description=IDStr(f"compressing {dir_to_compress.name}")
+            num_steps=1, description=f"compressing {dir_to_compress.name}"
         )
 
     options = " ".join(
@@ -223,7 +222,7 @@ async def archive_dir(
 
     async with AsyncExitStack() as exit_stack:
         sub_progress = await exit_stack.enter_async_context(
-            progress_bar.sub_progress(folder_size_bytes, description=IDStr("..."))
+            progress_bar.sub_progress(folder_size_bytes, description="...")
         )
 
         tqdm_progress = exit_stack.enter_context(
@@ -290,7 +289,7 @@ async def unarchive_dir(
 ) -> set[Path]:
     if progress_bar is None:
         progress_bar = ProgressBarData(
-            num_steps=1, description=IDStr(f"extracting {archive_to_extract.name}")
+            num_steps=1, description=f"extracting {archive_to_extract.name}"
         )
 
     # get archive information
@@ -304,7 +303,7 @@ async def unarchive_dir(
 
     async with AsyncExitStack() as exit_stack:
         sub_prog = await exit_stack.enter_async_context(
-            progress_bar.sub_progress(steps=total_bytes, description=IDStr("..."))
+            progress_bar.sub_progress(steps=total_bytes, description="...")
         )
 
         tqdm_progress = exit_stack.enter_context(

packages/service-library/src/servicelib/docker_utils.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from collections.abc import Awaitable, Callable
 from contextlib import AsyncExitStack
@@ -11,8 +12,21 @@
 from models_library.docker import DockerGenericTag
 from models_library.generated_models.docker_rest_api import ProgressDetail
 from models_library.utils.change_case import snake_to_camel
-from pydantic import BaseModel, ByteSize, ConfigDict, TypeAdapter, ValidationError
+from pydantic import (
+    BaseModel,
+    ByteSize,
+    ConfigDict,
+    NonNegativeInt,
+    TypeAdapter,
+    ValidationError,
+)
 from settings_library.docker_registry import RegistrySettings
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_random_exponential,
+)
 from yarl import URL
 
 from .logging_utils import LogLevelInt
@@ -209,6 +223,8 @@ async def pull_image(
     progress_bar: ProgressBarData,
     log_cb: LogCB,
     image_information: DockerImageManifestsV2 | None,
+    *,
+    retry_upon_error_count: NonNegativeInt = 10,
 ) -> None:
     """pull a docker image to the host machine.
 
@@ -219,7 +235,9 @@ async def pull_image(
         progress_bar -- the current progress bar
         log_cb -- a callback function to send logs to
         image_information -- the image layer information. If this is None, then no fine progress will be retrieved.
+        retry_upon_error_count -- number of tries if there is a TimeoutError. Usually cased by networking issues.
     """
+
     registry_auth = None
     if registry_settings.REGISTRY_URL and registry_settings.REGISTRY_URL in image:
         registry_auth = {
@@ -245,39 +263,64 @@ async def pull_image(
 
         client = await exit_stack.enter_async_context(aiodocker.Docker())
 
-        reported_progress = 0.0
-        async for pull_progress in client.images.pull(
-            image, stream=True, auth=registry_auth
-        ):
-            try:
-                parsed_progress = TypeAdapter(_DockerPullImage).validate_python(
-                    pull_progress
+        def _reset_progress_from_previous_attempt() -> None:
+            for pulled_status in layer_id_to_size.values():
+                pulled_status.downloaded = 0
+                pulled_status.extracted = 0
+
+        attempt: NonNegativeInt = 1
+
+        @retry(
+            wait=wait_random_exponential(),
+            stop=stop_after_attempt(retry_upon_error_count),
+            reraise=True,
+            retry=retry_if_exception_type(asyncio.TimeoutError),
+        )
+        async def _pull_image_with_retry() -> None:
+            nonlocal attempt
+            if attempt > 1:
+                # for each attempt rest the progress
+                progress_bar.reset()
+                _reset_progress_from_previous_attempt()
+            attempt += 1
+
+            _logger.info("attempt '%s' trying to pull image='%s'", attempt, image)
+
+            reported_progress = 0.0
+            async for pull_progress in client.images.pull(
+                image, stream=True, auth=registry_auth
+            ):
+                try:
+                    parsed_progress = TypeAdapter(_DockerPullImage).validate_python(
+                        pull_progress
+                    )
+                except ValidationError:
+                    _logger.exception(
+                        "Unexpected error while validating '%s'. "
+                        "TIP: This is probably an unforeseen pull status text that shall be added to the code. "
+                        "The pulling process will still continue.",
+                        f"{pull_progress=}",
+                    )
+                else:
+                    await _parse_pull_information(
+                        parsed_progress, layer_id_to_size=layer_id_to_size
+                    )
+
+                # compute total progress
+                total_downloaded_size = sum(
+                    layer.downloaded for layer in layer_id_to_size.values()
                 )
-            except ValidationError:
-                _logger.exception(
-                    "Unexpected error while validating '%s'. "
-                    "TIP: This is probably an unforeseen pull status text that shall be added to the code. "
-                    "The pulling process will still continue.",
-                    f"{pull_progress=}",
+                total_extracted_size = sum(
+                    layer.extracted for layer in layer_id_to_size.values()
                 )
-            else:
-                await _parse_pull_information(
-                    parsed_progress, layer_id_to_size=layer_id_to_size
+                total_progress = (total_downloaded_size + total_extracted_size) / 2.0
+                progress_to_report = total_progress - reported_progress
+                await progress_bar.update(progress_to_report)
+                reported_progress = total_progress
+
+                await log_cb(
+                    f"pulling {image_short_name}: {pull_progress}...",
+                    logging.DEBUG,
                 )
 
-            # compute total progress
-            total_downloaded_size = sum(
-                layer.downloaded for layer in layer_id_to_size.values()
-            )
-            total_extracted_size = sum(
-                layer.extracted for layer in layer_id_to_size.values()
-            )
-            total_progress = (total_downloaded_size + total_extracted_size) / 2.0
-            progress_to_report = total_progress - reported_progress
-            await progress_bar.update(progress_to_report)
-            reported_progress = total_progress
-
-            await log_cb(
-                f"pulling {image_short_name}: {pull_progress}...",
-                logging.DEBUG,
-            )
+        await _pull_image_with_retry()

packages/service-library/src/servicelib/fastapi/docker_utils.py

@@ -3,7 +3,6 @@
 from typing import Final
 
 import httpx
-from models_library.basic_types import IDStr
 from models_library.docker import DockerGenericTag
 from pydantic import ByteSize, TypeAdapter, ValidationError
 from settings_library.docker_registry import RegistrySettings
@@ -129,9 +128,7 @@ async def pull_images(
         num_steps=images_total_size,
         progress_report_cb=progress_cb,
         progress_unit="Byte",
-        description=IDStr(
-            f"pulling {len(images)} images",
-        ),
+        description=f"pulling {len(images)} images",
     ) as pbar:
 
         await asyncio.gather(

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

@@ -31,6 +31,7 @@ async def list_tasks(
 
 @router.get(
     "/{task_id}",
+    response_model=TaskStatus,
     responses={
         status.HTTP_404_NOT_FOUND: {"description": "Task does not exist"},
     },