♻️🔨 catalog: fixes mypy issues and add type-check step in CI

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

@@ -599,7 +599,10 @@ jobs:
         run: ./ci/helpers/show_system_versions.bash
       - name: install
         run: ./ci/github/unit-testing/catalog.bash install
+      - name: typecheck
+        run: ./ci/github/unit-testing/catalog.bash typecheck
       - name: test
+        if: always()
         run: ./ci/github/unit-testing/catalog.bash test
       - name: upload failed tests logs
         if: failure()

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

@@ -4,7 +4,7 @@
 from enum import Enum
 from functools import cached_property
 from pathlib import Path
-from typing import Any, Final, Iterator, Literal, Optional, Union
+from typing import Any, Final, Iterator, Literal, TypeAlias
 
 from pydantic import (
     BaseModel,
@@ -160,7 +160,7 @@ class PathMappingsLabel(BaseModel):
         description="optional list of paths which contents need to be persisted",
     )
 
-    state_exclude: Optional[set[str]] = Field(
+    state_exclude: set[str] | None = Field(
         None,
         description="optional list unix shell rules used to exclude files from the state",
     )
@@ -176,7 +176,7 @@ class Config(_BaseConfig):
         }
 
 
-ComposeSpecLabel = dict[str, Any]
+ComposeSpecLabel: TypeAlias = dict[str, Any]
 
 
 class RestartPolicy(str, Enum):
@@ -194,7 +194,7 @@ class _PortRange(BaseModel):
     @classmethod
     def lower_less_than_upper(cls, v, values) -> PortInt:
         upper = v
-        lower: Optional[PortInt] = values.get("lower")
+        lower: PortInt | None = values.get("lower")
         if lower is None or lower >= upper:
             raise ValueError(f"Condition not satisfied: {lower=} < {upper=}")
         return v
@@ -218,7 +218,7 @@ class Config(_BaseConfig):
 
 class NATRule(BaseModel):
     hostname: str
-    tcp_ports: list[Union[_PortRange, PortInt]]
+    tcp_ports: list[_PortRange | PortInt]
     dns_resolver: DNSResolver = Field(
         default_factory=lambda: DNSResolver(
             address=DEFAULT_DNS_SERVER_ADDRESS, port=DEFAULT_DNS_SERVER_PORT
@@ -235,7 +235,7 @@ def iter_tcp_ports(self) -> Iterator[PortInt]:
 
 
 class DynamicSidecarServiceLabels(BaseModel):
-    paths_mapping: Optional[Json[PathMappingsLabel]] = Field(
+    paths_mapping: Json[PathMappingsLabel] | None = Field(
         None,
         alias="simcore.service.paths-mapping",
         description=(
@@ -244,7 +244,7 @@ class DynamicSidecarServiceLabels(BaseModel):
         ),
     )
 
-    compose_spec: Optional[Json[ComposeSpecLabel]] = Field(
+    compose_spec: Json[ComposeSpecLabel] | None = Field(
         None,
         alias="simcore.service.compose-spec",
         description=(
@@ -253,7 +253,7 @@ class DynamicSidecarServiceLabels(BaseModel):
             "only used by dynamic-sidecar."
         ),
     )
-    container_http_entry: Optional[str] = Field(
+    container_http_entry: str | None = Field(
         None,
         alias="simcore.service.container-http-entrypoint",
         description=(
@@ -275,15 +275,15 @@ class DynamicSidecarServiceLabels(BaseModel):
         ),
     )
 
-    containers_allowed_outgoing_permit_list: Optional[
+    containers_allowed_outgoing_permit_list: None | (
         Json[dict[str, list[NATRule]]]
-    ] = Field(
+    ) = Field(
         None,
         alias="simcore.service.containers-allowed-outgoing-permit-list",
         description="allow internet access to certain domain names and ports per container",
     )
 
-    containers_allowed_outgoing_internet: Optional[Json[set[str]]] = Field(
+    containers_allowed_outgoing_internet: Json[set[str]] | None = Field(
         None,
         alias="simcore.service.containers-allowed-outgoing-internet",
         description="allow complete internet access to containers in here",
@@ -296,7 +296,7 @@ def needs_dynamic_sidecar(self) -> bool:
 
     @validator("container_http_entry", always=True)
     @classmethod
-    def compose_spec_requires_container_http_entry(cls, v, values) -> Optional[str]:
+    def compose_spec_requires_container_http_entry(cls, v, values) -> str | None:
         v = None if v == "" else v
         if v is None and values.get("compose_spec") is not None:
             raise ValueError(
@@ -314,7 +314,7 @@ def _containers_allowed_outgoing_permit_list_in_compose_spec(cls, v, values):
         if v is None:
             return v
 
-        compose_spec: Optional[dict] = values.get("compose_spec")
+        compose_spec: dict | None = values.get("compose_spec")
         if compose_spec is None:
             keys = set(v.keys())
             if len(keys) != 1 or DEFAULT_SINGLE_SERVICE_NAME not in keys:
@@ -337,7 +337,7 @@ def _containers_allowed_outgoing_internet_in_compose_spec(cls, v, values):
         if v is None:
             return v
 
-        compose_spec: Optional[dict] = values.get("compose_spec")
+        compose_spec: dict | None = values.get("compose_spec")
         if compose_spec is None:
             if {DEFAULT_SINGLE_SERVICE_NAME} != v:
                 raise ValueError(

services/catalog/src/simcore_service_catalog/api/routes/services_resources.py

@@ -1,7 +1,7 @@
 import logging
 import urllib.parse
 from copy import deepcopy
-from typing import Any, Final, Optional, cast
+from typing import Any, Final, cast
 
 import yaml
 from fastapi import APIRouter, Depends, HTTPException, status
@@ -59,7 +59,7 @@ def _compute_service_available_boot_modes(
     """
 
     resource_entries = filter(lambda entry: entry.name.lower() == "resources", settings)
-    generic_resources = {}
+    generic_resources: ResourcesDict = {}
     for entry in resource_entries:
         if not isinstance(entry.value, dict):
             logger.warning(
@@ -132,7 +132,7 @@ def _resources_from_settings(
 
 async def _get_service_labels(
     director_client: DirectorApi, key: ServiceKey, version: ServiceVersion
-) -> Optional[dict[str, Any]]:
+) -> dict[str, Any] | None:
     try:
         service_labels = cast(
             dict[str, Any],
@@ -186,7 +186,7 @@ async def get_service_resources(
             image_version, default_service_resources
         )
 
-    service_labels: Optional[dict[str, Any]] = await _get_service_labels(
+    service_labels: dict[str, Any] | None = await _get_service_labels(
         director_client, service_key, service_version
     )
 
@@ -195,8 +195,8 @@ async def get_service_resources(
             image_version, default_service_resources
         )
 
-    service_spec: Optional[ComposeSpecLabel] = parse_raw_as(
-        Optional[ComposeSpecLabel],
+    service_spec: ComposeSpecLabel | None = parse_raw_as(
+        ComposeSpecLabel | None,
         service_labels.get(SIMCORE_SERVICE_COMPOSE_SPEC_LABEL, "null"),
     )
     logger.debug("received %s", f"{service_spec=}")
@@ -243,16 +243,18 @@ async def get_service_resources(
         # leading slashes must be stripped
         image = spec_data["image"].lstrip("/")
         key, version = image.split(":")
-        spec_service_labels: Optional[dict[str, Any]] = await _get_service_labels(
+        spec_service_labels: dict[str, Any] | None = await _get_service_labels(
             director_client, key, version
         )
 
+        spec_service_resources: ResourcesDict
+
         if not spec_service_labels:
-            spec_service_resources: ResourcesDict = default_service_resources
+            spec_service_resources = default_service_resources
             service_boot_modes = [BootMode.CPU]
         else:
             spec_service_settings = _get_service_settings(spec_service_labels)
-            spec_service_resources: ResourcesDict = _resources_from_settings(
+            spec_service_resources = _resources_from_settings(
                 spec_service_settings,
                 default_service_resources,
                 service_key,

services/catalog/src/simcore_service_catalog/db/repositories/groups.py

@@ -36,7 +36,7 @@ async def get_everyone_group(self) -> GroupAtDB:
 
     async def get_user_gid_from_email(
         self, user_email: LowerCaseEmailStr
-    ) -> Optional[PositiveInt]:
+    ) -> PositiveInt | None:
         async with self.db_engine.connect() as conn:
             return cast(
                 Optional[PositiveInt],
@@ -45,7 +45,7 @@ async def get_user_gid_from_email(
                 ),
             )
 
-    async def get_gid_from_affiliation(self, affiliation: str) -> Optional[PositiveInt]:
+    async def get_gid_from_affiliation(self, affiliation: str) -> PositiveInt | None:
         async with self.db_engine.connect() as conn:
             return cast(
                 Optional[PositiveInt],
@@ -56,18 +56,16 @@ async def get_gid_from_affiliation(self, affiliation: str) -> Optional[PositiveI
 
     async def get_user_email_from_gid(
         self, gid: PositiveInt
-    ) -> Optional[LowerCaseEmailStr]:
+    ) -> LowerCaseEmailStr | None:
         async with self.db_engine.connect() as conn:
-            return cast(
-                Optional[LowerCaseEmailStr],
-                await conn.scalar(
-                    sa.select([users.c.email]).where(users.c.primary_gid == gid)
-                ),
+            email = await conn.scalar(
+                sa.select([users.c.email]).where(users.c.primary_gid == gid)
             )
+            return cast(LowerCaseEmailStr, f"{email}") if email else None
 
     async def list_user_emails_from_gids(
         self, gids: set[PositiveInt]
-    ) -> dict[PositiveInt, Optional[LowerCaseEmailStr]]:
+    ) -> dict[PositiveInt, LowerCaseEmailStr | None]:
         service_owners = {}
         async with self.db_engine.connect() as conn:
             async for row in await conn.stream(

services/catalog/src/simcore_service_catalog/models/domain/dag.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from models_library.basic_regex import VERSION_RE
 from models_library.emails import LowerCaseEmailStr
 from models_library.projects_nodes import Node
@@ -15,17 +13,18 @@ class DAGBase(BaseModel):
     )
     version: str = Field(..., regex=VERSION_RE, example="1.0.0")
     name: str
-    description: Optional[str]
-    contact: Optional[LowerCaseEmailStr]
+    description: str | None
+    contact: LowerCaseEmailStr | None
 
 
 class DAGAtDB(DAGBase):
     id: int
-    workbench: Json[dict[str, Node]]  # pylint: disable=unsubscriptable-object
+    # pylint: disable=unsubscriptable-object
+    workbench: Json[dict[str, Node]]  # type: ignore
 
     class Config:
         orm_mode = True
 
 
 class DAGData(DAGAtDB):
-    workbench: Optional[dict[str, Node]]
+    workbench: dict[str, Node] | None  # type: ignore