♻️ Maintenance: mypy dynamic-sidecar (#6126)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

.github/workflows/ci-testing-deploy.yml

@@ -1212,7 +1212,6 @@ jobs:
         run: ./ci/github/unit-testing/dynamic-sidecar.bash install
       - name: typecheck
         run: ./ci/github/unit-testing/dynamic-sidecar.bash typecheck
-        continue-on-error: true
       - name: test
         if: always()
         run: ./ci/github/unit-testing/dynamic-sidecar.bash test

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/io.py

@@ -3,12 +3,12 @@
 from pathlib import Path
 from typing import Any, ClassVar, TypeAlias, Union
 
-from models_library.basic_regex import MIME_TYPE_RE, PROPERTY_KEY_RE
+from models_library.basic_regex import MIME_TYPE_RE
 from models_library.generics import DictModel
+from models_library.services_types import ServicePortKey
 from pydantic import (
     AnyUrl,
     BaseModel,
-    ConstrainedStr,
     Extra,
     Field,
     StrictBool,
@@ -81,10 +81,6 @@ class Config:
         }
 
 
-class PortKey(ConstrainedStr):
-    regex = PROPERTY_KEY_RE
-
-
 PortValue: TypeAlias = Union[
     StrictBool,
     StrictInt,
@@ -97,7 +93,7 @@ class PortKey(ConstrainedStr):
 ]
 
 
-class TaskInputData(DictModel[PortKey, PortValue]):
+class TaskInputData(DictModel[ServicePortKey, PortValue]):
     class Config:
         schema_extra: ClassVar[dict[str, Any]] = {
             "examples": [
@@ -115,7 +111,7 @@ class Config:
 PortSchemaValue: TypeAlias = Union[PortSchema, FilePortSchema]
 
 
-class TaskOutputDataSchema(DictModel[PortKey, PortSchemaValue]):
+class TaskOutputDataSchema(DictModel[ServicePortKey, PortSchemaValue]):
     #
     # NOTE: Expected output data is only determined at runtime. A possibility
     # would be to create pydantic models dynamically but dask serialization
@@ -144,7 +140,7 @@ class Config:
         }
 
 
-class TaskOutputData(DictModel[PortKey, PortValue]):
+class TaskOutputData(DictModel[ServicePortKey, PortValue]):
     @classmethod
     def from_task_output(
         cls, schema: TaskOutputDataSchema, output_folder: Path, output_file_ext: str

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

@@ -9,7 +9,10 @@
 from typing import Any, Protocol
 from uuid import uuid4
 
-from models_library.api_schemas_long_running_tasks.base import ProgressPercent
+from models_library.api_schemas_long_running_tasks.base import (
+    ProgressPercent,
+    TaskProgress,
+)
 from pydantic import PositiveFloat
 
 from ._errors import (
@@ -19,7 +22,7 @@
     TaskNotCompletedError,
     TaskNotFoundError,
 )
-from ._models import TaskId, TaskName, TaskProgress, TaskResult, TaskStatus, TrackedTask
+from ._models import TaskId, TaskName, TaskResult, TaskStatus, TrackedTask
 
 logger = logging.getLogger(__name__)
 
@@ -349,8 +352,7 @@ async def close(self) -> None:
 
 
 class TaskProtocol(Protocol):
-    # NOTE: when using **kwargs pyright complains. this might be a bug that should be fixed soon
-    async def __call__(self, task_progress: TaskProgress, *task_kwargs: Any) -> Any:
+    async def __call__(self, progress: TaskProgress, *args: Any, **kwargs: Any) -> Any:
         ...
 
     @property
@@ -366,7 +368,7 @@ def start_task(
     task_context: TaskContext | None = None,
     task_name: str | None = None,
     fire_and_forget: bool = False,
-    **task_kwargs,
+    **task_kwargs: Any,
 ) -> TaskId:
     """
     Creates a background task from an async function.

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/nodeports_v2.py

@@ -1,11 +1,13 @@
 import logging
+from collections.abc import Callable, Coroutine
 from pathlib import Path
-from typing import Any, Callable, Coroutine
+from typing import Any
 
 from models_library.api_schemas_storage import LinkType
 from models_library.basic_types import IDStr
 from models_library.projects import ProjectIDStr
 from models_library.projects_nodes_io import NodeIDStr
+from models_library.services_types import ServicePortKey
 from models_library.users import UserID
 from pydantic import BaseModel, Field, ValidationError
 from pydantic.error_wrappers import flatten_errors
@@ -20,7 +22,7 @@
 from ..node_ports_v2.port import SetKWargs
 from .links import ItemConcreteValue, ItemValue
 from .port_utils import is_file_type
-from .ports_mapping import InputsList, OutputsList, PortKey
+from .ports_mapping import InputsList, OutputsList
 
 log = logging.getLogger(__name__)
 
@@ -74,7 +76,7 @@ async def outputs(self) -> OutputsList:
         return self.internal_outputs
 
     async def get_value_link(
-        self, item_key: PortKey, *, file_link_type: LinkType
+        self, item_key: ServicePortKey, *, file_link_type: LinkType
     ) -> ItemValue | None:
         try:
             return await (await self.inputs)[item_key].get_value(
@@ -89,7 +91,7 @@ async def get_value_link(
         )
 
     async def get(
-        self, item_key: PortKey, progress_bar: ProgressBarData | None = None
+        self, item_key: ServicePortKey, progress_bar: ProgressBarData | None = None
     ) -> ItemConcreteValue | None:
         try:
             return await (await self.inputs)[item_key].get(progress_bar)
@@ -99,7 +101,9 @@ async def get(
         # if this fails it will raise an exception
         return await (await self.outputs)[item_key].get(progress_bar)
 
-    async def set(self, item_key: PortKey, item_value: ItemConcreteValue) -> None:
+    async def set(
+        self, item_key: ServicePortKey, item_value: ItemConcreteValue
+    ) -> None:
         # first try to set the inputs.
         try:
             the_updated_inputs = await self.inputs
@@ -139,7 +143,9 @@ async def _auto_update_from_db(self) -> None:
 
     async def set_multiple(
         self,
-        port_values: dict[PortKey, tuple[ItemConcreteValue | None, SetKWargs | None]],
+        port_values: dict[
+            ServicePortKey, tuple[ItemConcreteValue | None, SetKWargs | None]
+        ],
         *,
         progress_bar: ProgressBarData,
     ) -> None:

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/port.py

@@ -7,9 +7,9 @@
 from typing import Any
 
 from models_library.api_schemas_storage import LinkType
-from models_library.basic_regex import PROPERTY_KEY_RE
 from models_library.basic_types import IDStr
 from models_library.services_io import BaseServiceIOModel
+from models_library.services_types import ServicePortKey
 from pydantic import AnyUrl, Field, PrivateAttr, ValidationError, validator
 from pydantic.tools import parse_obj_as
 from servicelib.progress_bar import ProgressBarData
@@ -68,7 +68,7 @@ class SetKWargs:
 
 
 class Port(BaseServiceIOModel):
-    key: str = Field(..., regex=PROPERTY_KEY_RE)
+    key: ServicePortKey
     widget: dict[str, Any] | None = None
     default_value: DataItemValue | None = Field(None, alias="defaultValue")
 

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/ports_mapping.py

@@ -1,21 +1,16 @@
-import re
 from collections.abc import ItemsView, Iterator, KeysView, ValuesView
 
-from models_library.basic_regex import PROPERTY_KEY_RE
-from pydantic import BaseModel, ConstrainedStr
+from models_library.services_types import ServicePortKey
+from pydantic import BaseModel
 
 from ..node_ports_common.exceptions import UnboundPortError
 from .port import Port
 
 
-class PortKey(ConstrainedStr):
-    regex = re.compile(PROPERTY_KEY_RE)
-
-
 class BasePortsMapping(BaseModel):
-    __root__: dict[PortKey, Port]
+    __root__: dict[ServicePortKey, Port]
 
-    def __getitem__(self, key: int | PortKey) -> Port:
+    def __getitem__(self, key: int | ServicePortKey) -> Port:
         if isinstance(key, int):
             if key < len(self.__root__):
                 key = list(self.__root__.keys())[key]
@@ -24,13 +19,13 @@ def __getitem__(self, key: int | PortKey) -> Port:
         assert isinstance(key, str)  # nosec
         return self.__root__[key]
 
-    def __iter__(self) -> Iterator[PortKey]:  # type: ignore
+    def __iter__(self) -> Iterator[ServicePortKey]:  # type: ignore
         return iter(self.__root__)
 
-    def keys(self) -> KeysView[PortKey]:
+    def keys(self) -> KeysView[ServicePortKey]:
         return self.__root__.keys()
 
-    def items(self) -> ItemsView[PortKey, Port]:
+    def items(self) -> ItemsView[ServicePortKey, Port]:
         return self.__root__.items()
 
     def values(self) -> ValuesView[Port]: