mypy

Author: sanderegg

packages/models-library/src/models_library/function_services_catalog/_utils.py

@@ -52,7 +52,7 @@ def add(
         raises ValueError
         """
         if not isinstance(meta, ServiceMetaDataPublished):
-            msg = f"Expected ServiceDockerData, got {type(meta)}"
+            msg = f"Expected ServiceDockerData, got {type(meta)}"  # type: ignore[unreachable]
             raise ValueError(msg)
 
         # ensure unique

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

@@ -1,4 +1,4 @@
-from typing import Literal, TypeAlias
+from typing import Final, Literal, TypeAlias
 
 from pydantic import BaseModel, ConfigDict
 
@@ -43,7 +43,7 @@ class ProgressStructuredMessage(BaseModel):
     )
 
 
-UNITLESS = None
+UNITLESS: Final[None] = None
 
 
 class ProgressReport(BaseModel):
@@ -96,7 +96,11 @@ def composed_message(self) -> str:
                 {
                     "actual_value": 0.3,
                     "total": 1.0,
-                    "message": ProgressStructuredMessage.model_config["json_schema_extra"]["examples"][2],  # type: ignore [index]
+                    "message": ProgressStructuredMessage.model_config[
+                        "json_schema_extra"
+                    ]["examples"][
+                        2
+                    ],  # type: ignore [index]
                 },
             ]
         },

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

@@ -518,7 +518,7 @@ def _ensure_callbacks_mapping_container_names_defined_in_compose_spec(
         cls, v: CallbacksMapping, info: ValidationInfo
     ):
         if v is None:
-            return {}
+            return {}  # type: ignore[unreachable]
 
         defined_services: set[str] = {x.service for x in v.before_shutdown}
         if v.metrics:

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

@@ -97,5 +97,5 @@ def validate(cls, v: "ServiceRunID | str", _: ValidationInfo) -> "ServiceRunID":
             return v
         if isinstance(v, str):
             return cls(v)
-        msg = f"Invalid value for RunID: {v}"
+        msg = f"Invalid value for RunID: {v}"  # type: ignore[unreachable]
         raise TypeError(msg)

packages/models-library/src/models_library/utils/_original_fastapi_encoders.py

@@ -5,15 +5,16 @@
 #
 import dataclasses
 from collections import defaultdict, deque
+from collections.abc import Callable
 from enum import Enum
 from pathlib import PurePath
 from types import GeneratorType
-from typing import Any, Callable, Union, get_origin
+from typing import Annotated, Any, Union, get_origin
 
 from common_library.json_serialization import ENCODERS_BY_TYPE
 from pydantic import BaseModel
 from pydantic_core import PydanticUndefined, PydanticUndefinedType
-from typing_extensions import Annotated, Doc
+from typing_extensions import Doc
 
 Undefined = PydanticUndefined
 UndefinedType = PydanticUndefinedType
@@ -142,14 +143,13 @@ def jsonable_encoder(
     if custom_encoder:
         if type(obj) in custom_encoder:
             return custom_encoder[type(obj)](obj)
-        else:
-            for encoder_type, encoder_instance in custom_encoder.items():
-                if isinstance(obj, encoder_type):
-                    return encoder_instance(obj)
-    if include is not None and not isinstance(include, (set, dict)):
-        include = set(include)
-    if exclude is not None and not isinstance(exclude, (set, dict)):
-        exclude = set(exclude)
+        for encoder_type, encoder_instance in custom_encoder.items():
+            if isinstance(obj, encoder_type):
+                return encoder_instance(obj)
+    if include is not None and not isinstance(include, set | dict):
+        include = set(include)  # type: ignore[unreachable]
+    if exclude is not None and not isinstance(exclude, set | dict):
+        exclude = set(exclude)  # type: ignore[unreachable]
     if isinstance(obj, BaseModel):
         obj_dict = BaseModel.model_dump(
             obj,

packages/models-library/src/models_library/utils/common_validators.py

@@ -27,7 +27,7 @@ class MyModel(BaseModel):
 
 
 def trim_string_before(max_length: int) -> BeforeValidator:
-    def _trim(value: str):
+    def _trim(value: str | Any) -> str | Any:
         if isinstance(value, str):
             return value[:max_length]
         return value
@@ -125,14 +125,16 @@ def _validator(cls: type[BaseModel], values):
         }
 
         if not functools.reduce(operator.xor, (v is not None for v in got.values())):
-            msg = f"Either { ' or '.join(got.keys()) } must be set, but not both. Got {got}"
+            msg = (
+                f"Either {' or '.join(got.keys())} must be set, but not both. Got {got}"
+            )
             raise ValueError(msg)
         return values
 
     return _validator
 
 
-def to_camel_recursive(data: dict[str, Any]) -> dict[str, Any]:
+def to_camel_recursive(data: dict[str, Any] | Any) -> dict[str, Any] | Any:
     """Recursively convert dictionary keys to camelCase"""
     if not isinstance(data, dict):
         return data  # Return as-is if it's not a dictionary