Upgrading mypy

pyproject.toml

@@ -121,7 +121,7 @@ dev = [
     "yamlfix>=1.16.0",
     "coverage[toml]>=5.5,<6.0.0",
     "pytest>=7.4.0,<8.0.0",
-    "mypy==1.7.1",
+    "mypy==1.18.1",
     "pre-commit",
     "pyment>=0.3.3,<0.4.0",
     "tox>=3.24.3",

src/zenml/integrations/bentoml/model_deployers/bentoml_model_deployer.py

@@ -142,11 +142,11 @@ def get_model_server_info(
             )
 
         predictions_apis_urls = ""
-        if service_instance.prediction_apis_urls is not None:  # type: ignore
+        if service_instance.prediction_apis_urls is not None:
             predictions_apis_urls = ", ".join(
                 [
                     api
-                    for api in service_instance.prediction_apis_urls  # type: ignore
+                    for api in service_instance.prediction_apis_urls
                     if api is not None
                 ]
             )

src/zenml/integrations/databricks/orchestrators/databricks_orchestrator_entrypoint_config.py

@@ -82,7 +82,9 @@ def run(self) -> None:
         wheel_package = self.entrypoint_args[WHEEL_PACKAGE_OPTION]
 
         dist = distribution(wheel_package)
-        project_root = os.path.join(dist.locate_file("."), wheel_package)
+        project_root = os.path.join(
+            str(dist.locate_file(".")), str(wheel_package)
+        )
 
         if project_root not in sys.path:
             sys.path.insert(0, project_root)

src/zenml/integrations/kubernetes/orchestrators/kubernetes_orchestrator_entrypoint.py

@@ -158,8 +158,10 @@ def _reconstruct_nodes(
     )
     for job in job_list.items:
         annotations = job.metadata.annotations or {}
-        if step_name := annotations.get(STEP_NAME_ANNOTATION_KEY, None):
-            node = nodes[step_name]
+        if step_name_annotation_key := annotations.get(
+            STEP_NAME_ANNOTATION_KEY, None
+        ):
+            node = nodes[str(step_name_annotation_key)]
             node.metadata["job_name"] = job.metadata.name
 
             if node.status == NodeStatus.NOT_READY:

src/zenml/integrations/seldon/services/seldon_deployment.py

@@ -72,7 +72,7 @@ class SeldonDeploymentConfig(ServiceConfig):
     model_metadata: Dict[str, Any] = Field(default_factory=dict)
     extra_args: Dict[str, Any] = Field(default_factory=dict)
     is_custom_deployment: Optional[bool] = False
-    spec: Optional[Dict[Any, Any]] = Field(default_factory=dict)  # type: ignore[arg-type]
+    spec: Optional[Dict[Any, Any]] = Field(default_factory=dict)
     serviceAccountName: Optional[str] = None
 
     def get_seldon_deployment_labels(self) -> Dict[str, str]:

src/zenml/models/v2/base/filter.py

@@ -290,7 +290,7 @@ def _handle_numeric_comparison(self, column: Any) -> Any:
                 ),
             }
 
-            return operations[self.operation](numeric_column, self.value)  # type: ignore[no-untyped-call]
+            return operations[self.operation](numeric_column, self.value)
         except Exception as e:
             raise ValueError(
                 f"Failed to compare the column '{column}' to the "
@@ -990,7 +990,7 @@ def apply_sorting(
         column, operand = self.sorting_params
 
         if operand == SorterOps.DESCENDING:
-            sort_clause = desc(getattr(table, column))  # type: ignore[var-annotated]
+            sort_clause = desc(getattr(table, column))
         else:
             sort_clause = asc(getattr(table, column))
 

src/zenml/utils/function_utils.py

@@ -143,7 +143,7 @@ def _cli_wrapped_function(func: F) -> F:
     )
     invalid_types = {}
     for arg_name, arg_type, arg_default in input_args_dict:
-        if _is_valid_optional_arg(arg_type):
+        if arg_type is not None and _is_valid_optional_arg(arg_type):
             arg_type = arg_type.__args__[0]
         arg_name = _cli_arg_name(arg_name)
         if arg_type is bool:
@@ -156,7 +156,7 @@ def _cli_wrapped_function(func: F) -> F:
                     required=False,
                 )
             )
-        elif _is_valid_collection_arg(arg_type):
+        elif arg_type is not None and _is_valid_collection_arg(arg_type):
             member_type = arg_type.__args__[0]
             options.append(
                 click.option(

src/zenml/utils/json_utils.py

@@ -114,7 +114,7 @@ def pydantic_encoder(obj: Any) -> Any:
 
     if isinstance(obj, BaseModel):
         return obj.model_dump(mode="json")
-    elif is_dataclass(obj):
+    elif is_dataclass(obj) and not isinstance(obj, type):
         return asdict(obj)
 
     # Check the class type and its superclasses for a matching encoder

src/zenml/utils/proxy_utils.py

@@ -207,6 +207,6 @@ def _inner_decorator(_cls: C) -> C:
             if method_name not in interface.__abstractmethods__
         )
 
-        return cast(C, _cls)
+        return cast(C, _cls)  # type: ignore[redundant-cast]
 
     return _inner_decorator

src/zenml/utils/string_utils.py

@@ -232,7 +232,7 @@ def substitute_string(value: V, substitution_func: Callable[[str], str]) -> V:
 
             model_values[k] = new_value
 
-        return cast(V, type(value).model_validate(model_values))
+        return cast(V, type(value).model_validate(model_values))  # type: ignore[redundant-cast]
     elif isinstance(value, Dict):
         return cast(
             V, {substitute_(k): substitute_(v) for k, v in value.items()}

src/zenml/utils/typing_utils.py

@@ -176,7 +176,7 @@ def _generic_get_args(tp: Type[Any]) -> Tuple[Any, ...]:
     # Special case for `tuple[()]`, which used to return ((),) with
     # `typing.Tuple in python 3.10- but now returns () for `tuple` and `Tuple`.
     try:
-        if tp == Tuple[()] or sys.version_info >= (3, 9) and tp == tuple[()]:
+        if tp == Tuple[()] or tp == tuple[()]:  # type: ignore[comparison-overlap]
             return ((),)
     # there is a TypeError when compiled with cython
     except TypeError:  # pragma: no cover

src/zenml/zen_server/middleware.py

@@ -111,7 +111,7 @@ async def dispatch(
 class RestrictFileUploadsMiddleware(BaseHTTPMiddleware):
     """Restrict file uploads to certain paths."""
 
-    def __init__(self, app: FastAPI, allowed_paths: Set[str]):
+    def __init__(self, app: ASGIApp, allowed_paths: Set[str]):
         """Restrict file uploads to certain paths.
 
         Args:
@@ -404,7 +404,7 @@ def add_middlewares(app: FastAPI) -> None:
 
     app.add_middleware(
         CORSMiddleware,
-        allow_origins=server_config().cors_allow_origins,
+        allow_origins=server_config().cors_allow_origins or [],
         allow_credentials=True,
         allow_methods=["*"],
         allow_headers=["*"],

src/zenml/zen_stores/sql_zen_store.py

@@ -10718,7 +10718,7 @@ def _get_resource_references(
         for resource_attr in resource_attrs:
             # Extract the target schema from the annotation
             annotation = UserSchema.__annotations__[resource_attr]
-            if get_origin(annotation) == Mapped:
+            if get_origin(annotation) == Mapped:  # type: ignore[comparison-overlap]
                 annotation = annotation.__args__[0]
 
             # The annotation must be of the form
@@ -10736,8 +10736,10 @@ def _get_resource_references(
                     vars(zenml_schemas), {}, recursive_guard=frozenset()
                 )
             else:
-                target_schema = schema_ref._evaluate(
-                    vars(zenml_schemas), {}, frozenset()
+                target_schema = schema_ref._evaluate(  # type: ignore[unused-ignore, call-arg, call-overload]
+                    vars(zenml_schemas),
+                    {},
+                    frozenset(),  # type: ignore[unused-ignore, arg-type]
                 )
             assert target_schema is not None
             assert issubclass(target_schema, SQLModel)

src/zenml/zen_stores/template_utils.py

@@ -199,7 +199,7 @@ def generate_config_schema(
         ]:
             continue
 
-        if field_info.annotation == Optional[SourceWithValidator]:
+        if field_info.annotation == Optional[SourceWithValidator]:  # type: ignore[comparison-overlap]
             generic_step_fields[key] = (Optional[str], field_info)
         else:
             generic_step_fields[key] = (field_info.annotation, field_info)
@@ -279,7 +279,7 @@ def generate_config_schema(
         if key in ["schedule", "build", "steps", "settings", "parameters"]:
             continue
 
-        if field_info.annotation == Optional[SourceWithValidator]:
+        if field_info.annotation == Optional[SourceWithValidator]:  # type: ignore[comparison-overlap]
             top_level_fields[key] = (Optional[str], field_info)
         else:
             top_level_fields[key] = (field_info.annotation, field_info)