mypy

Author: Andrei Neagu

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rest/containers.py

@@ -241,7 +241,9 @@ async def get_container_logs(
         if timestamps:
             args["timestamps"] = True
 
-        container_logs: list[str] = await container_instance.log(**args)
+        container_logs: list[str] = await container_instance.log(
+            **args
+        )  # type:ignore[call-overload]
         return container_logs
 
 

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

@@ -1,7 +1,7 @@
 import logging
 from collections.abc import AsyncGenerator, Iterable
 from contextlib import asynccontextmanager
-from typing import Any, cast
+from typing import Any
 
 import aiodocker
 import yaml
@@ -65,7 +65,7 @@ async def _get_container(
     docker: aiodocker.Docker, container_name: str
 ) -> DockerContainer | None:
     try:
-        return cast(DockerContainer, await docker.containers.get(container_name))
+        return await docker.containers.get(container_name)
     except aiodocker.DockerError as e:
         if e.status == http_status.HTTP_404_NOT_FOUND:
             return None

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/attribute_monitor/_logging_event_handler.py

@@ -27,7 +27,11 @@ class _LoggingEventHandler(SafeFileSystemEventHandler):
     def event_handler(self, event: FileSystemEvent) -> None:
         # NOTE: runs in the created process
 
-        file_path = Path(event.src_path)
+        file_path = Path(
+            event.src_path.decode()
+            if isinstance(event.src_path, bytes)
+            else event.src_path
+        )
         with suppress(FileNotFoundError):
             file_stat = file_path.stat()
             logger.info(

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/attribute_monitor/_watchdog_extensions.py

@@ -15,13 +15,15 @@
 
 
 class _ExtendedInotifyBuffer(InotifyBuffer):
-    def __init__(self, path, recursive=False):  # pylint:disable=super-init-not-called
+    def __init__(
+        self, path: bytes, *, recursive: bool = False
+    ):  # pylint:disable=super-init-not-called
         # below call to `BaseThread.__init__` is correct since we want to
         # overwrite the `InotifyBuffer.__init__` method
         BaseThread.__init__(self)  # pylint:disable=non-parent-init-called
         self._queue = DelayedQueue(self.delay)
         self._inotify = Inotify(  # pylint:disable=too-many-function-args
-            path, recursive, InotifyConstants.IN_ATTRIB
+            path, recursive=recursive, event_mask=InotifyConstants.IN_ATTRIB
         )
         self.start()
 
@@ -30,7 +32,7 @@ class _ExtendedInotifyEmitter(InotifyEmitter):
     def on_thread_start(self):
         path = os.fsencode(self.watch.path)
         # pylint:disable=attribute-defined-outside-init
-        self._inotify = _ExtendedInotifyBuffer(path, self.watch.is_recursive)
+        self._inotify = _ExtendedInotifyBuffer(path, recursive=self.watch.is_recursive)
 
 
 class ExtendedInotifyObserver(BaseObserver):

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/outputs/_event_handler.py

@@ -49,7 +49,11 @@ def event_handler(self, event: FileSystemEvent) -> None:
         # NOTE: ignoring all events which are not relative to modifying
         # the contents of the `port_key` folders from the outputs directory
 
-        path_relative_to_outputs = Path(event.src_path).relative_to(self.outputs_path)
+        path_relative_to_outputs = Path(
+            event.src_path.decode()
+            if isinstance(event.src_path, bytes)
+            else event.src_path
+        ).relative_to(self.outputs_path)
 
         # discard event if not part of a subfolder
         relative_path_parents = path_relative_to_outputs.parents

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/outputs/_watchdog_extensions.py

@@ -31,13 +31,15 @@
 
 
 class _ExtendedInotifyBuffer(InotifyBuffer):
-    def __init__(self, path, recursive=False):  # pylint:disable=super-init-not-called
+    def __init__(
+        self, path: bytes, *, recursive: bool = False
+    ):  # pylint:disable=super-init-not-called
         # below call to `BaseThread.__init__` is correct since we want to
         # overwrite the `InotifyBuffer.__init__` method
         BaseThread.__init__(self)  # pylint:disable=non-parent-init-called
         self._queue = DelayedQueue(self.delay)
         self._inotify = Inotify(  # pylint:disable=too-many-function-args
-            path, recursive, _EVENTS_TO_WATCH
+            path, recursive=recursive, event_mask=_EVENTS_TO_WATCH
         )
         self.start()
 
@@ -46,7 +48,7 @@ class _ExtendedInotifyEmitter(InotifyEmitter):
     def on_thread_start(self):
         path = os.fsencode(self.watch.path)
         # pylint:disable=attribute-defined-outside-init
-        self._inotify = _ExtendedInotifyBuffer(path, self.watch.is_recursive)
+        self._inotify = _ExtendedInotifyBuffer(path, recursive=self.watch.is_recursive)
 
 
 class ExtendedInotifyObserver(BaseObserver):