refactored errors

Author: Andrei Neagu

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/docker_utils.py

@@ -50,7 +50,13 @@ async def get_volume_by_label(label: str, run_id: RunID) -> dict[str, Any]:
         volumes = data["Volumes"]
         _logger.debug("volumes query for label=%s volumes=%s", label, volumes)
         if len(volumes) != 1:
-            raise VolumeNotFoundError(label, run_id, volumes)
+            raise VolumeNotFoundError(
+                volume_count=len(volumes),
+                source_label=label,
+                run_id=run_id,
+                volume_names=" ".join(v.get("Name", "UNKNOWN") for v in volumes),
+                status_code=http_status.HTTP_404_NOT_FOUND,
+            )
         volume_details: dict[str, Any] = volumes[0]
         return volume_details
 

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/error_handlers.py

@@ -12,7 +12,7 @@ async def http_error_handler(
 ) -> JSONResponse:
     return JSONResponse(
         content=jsonable_encoder({"errors": [exception.message]}),
-        status_code=exception.status_code,
+        status_code=exception.status_code,  # type:ignore[attr-defined]
     )
 
 

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/errors.py

@@ -1,53 +1,30 @@
-from typing import Any
-
 from common_library.errors_classes import OsparcErrorMixin
-from fastapi import status
-from models_library.services import RunID
 
 
-class BaseDynamicSidecarError(Exception):
+class BaseDynamicSidecarError(OsparcErrorMixin, Exception):
     """Used as base for all exceptions"""
 
-    def __init__(
-        self, nessage: str, status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR
-    ) -> None:
-        self.message: str = nessage
-        self.status_code: int = status_code
-        super().__init__(nessage)
-
 
 class VolumeNotFoundError(BaseDynamicSidecarError):
-    def __init__(
-        self, source_label: str, run_id: RunID, volumes: list[dict[str, Any]]
-    ) -> None:
-        super().__init__(
-            f"Expected 1 got {len(volumes)} volumes labels with {source_label=}, {run_id=}: "
-            f"Found {' '.join(v.get('Name', 'UNKNOWN') for v in volumes)}",
-            status_code=status.HTTP_404_NOT_FOUND,
-        )
+    msg_template = (
+        "Expected 1 got {volume_count} volumes labels with "
+        "source_label={source_label}, run_id={run_id}: Found {volume_names}"
+    )
 
 
 class UnexpectedDockerError(BaseDynamicSidecarError):
-    def __init__(self, message: str, status_code: int) -> None:
-        super().__init__(
-            f"An unexpected Docker error occurred {status_code=}, {message=}",
-            status_code=status_code,
-        )
-
-
-class BaseError(OsparcErrorMixin, BaseDynamicSidecarError):
-    ...
+    msg_template = "An unexpected Docker error occurred status_code={status_code}, message={message}"
 
 
-class ContainerExecContainerNotFoundError(BaseError):
+class ContainerExecContainerNotFoundError(BaseDynamicSidecarError):
     msg_template = "Container '{container_name}' was not found"
 
 
-class ContainerExecTimeoutError(BaseError):
+class ContainerExecTimeoutError(BaseDynamicSidecarError):
     msg_template = "Timed out after {timeout} while executing: '{command}'"
 
 
-class ContainerExecCommandFailedError(BaseError):
+class ContainerExecCommandFailedError(BaseDynamicSidecarError):
     msg_template = (
         "Command '{command}' exited with code '{exit_code}'"
         "and output: '{command_result}'"

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/settings.py

@@ -2,7 +2,7 @@
 from datetime import timedelta
 from functools import lru_cache
 from pathlib import Path
-from typing import cast
+from typing import Annotated, cast
 
 from common_library.pydantic_validators import validate_numeric_string_as_timedelta
 from models_library.basic_types import BootModeEnum, PortInt
@@ -53,10 +53,13 @@ class SystemMonitorSettings(BaseCustomSettings):
 
 
 class ApplicationSettings(BaseCustomSettings, MixinLoggingSettings):
-    SC_BOOT_MODE: BootModeEnum = Field(
-        ...,
-        description="boot mode helps determine if in development mode or normal operation",
-    )
+    SC_BOOT_MODE: Annotated[
+        BootModeEnum,
+        Field(
+            ...,
+            description="boot mode helps determine if in development mode or normal operation",
+        ),
+    ]
 
     DYNAMIC_SIDECAR_DY_VOLUMES_MOUNT_DIR: Path = Field(
         ...,

services/dynamic-sidecar/tests/unit/test_core_errors.py

@@ -23,13 +23,20 @@ def test_legacy_interface_unexpected_docker_error():
 
 def test_legacy_interface_volume_not_found_error():
     try:
+        volumes = [{}, {"Name": "a_volume"}]
+        volume_names = " ".join(v.get("Name", "UNKNOWN") for v in volumes)
+
         raise VolumeNotFoundError(  # noqa: TRY301
-            source_label="some", run_id="run_id", volumes=[{}, {"Name": "a_volume"}]
+            volume_count=len(volumes),
+            source_label="some",
+            run_id="run_id",
+            volume_names=volume_names,
+            status_code=status.HTTP_404_NOT_FOUND,
         )
     except Exception as e:
         print(e)
         assert (  # noqa: PT017
             e.message
-            == "Expected 1 got 2 volumes labels with source_label='some', run_id='run_id': Found UNKNOWN a_volume"
+            == "Expected 1 got 2 volumes labels with source_label=some, run_id=run_id: Found UNKNOWN a_volume"
         )
         assert e.status_code == status.HTTP_404_NOT_FOUND  # noqa: PT017