Enable more strict mypy rules

ixmp4/cli/platforms.py

@@ -2,7 +2,7 @@
 from collections.abc import Generator, Iterator
 from itertools import cycle
 from pathlib import Path
-from typing import Optional
+from typing import Any, Optional, TypeVar
 
 import typer
 from rich.progress import Progress, track
@@ -272,7 +272,12 @@ def generate(
         utils.good("Done!")
 
 
-def create_cycle(generator: Generator, name: str, total: int) -> Iterator:
+T = TypeVar("T")
+
+
+def create_cycle(
+    generator: Generator[T, Any, None], name: str, total: int
+) -> Iterator[T]:
     return cycle(
         [
             m

ixmp4/conf/auth.py

@@ -1,7 +1,7 @@
 import logging
 from collections.abc import Generator
 from datetime import datetime, timedelta
-from typing import Any
+from typing import Any, cast
 from uuid import uuid4
 
 import httpx
@@ -161,11 +161,13 @@ def refresh_jwt(self) -> None:
         self.set_user(self.access_token)
 
     def decode_token(self, token: str) -> dict[str, Any]:
-        decoded: dict = jwt.decode(
-            token,
-            options={"verify_signature": False, "verify_exp": False},
+        return cast(
+            dict[str, Any],
+            jwt.decode(
+                token,
+                options={"verify_signature": False, "verify_exp": False},
+            ),
         )
-        return decoded
 
     def set_user(self, token: str) -> None:
         token_dict = self.decode_token(token)

ixmp4/conf/credentials.py

@@ -5,7 +5,7 @@
 
 
 class Credentials(object):
-    credentials: dict
+    credentials: dict[str, dict[str, str]]
 
     def __init__(self, toml_file: Path) -> None:
         self.path = toml_file

ixmp4/conf/manager.py

@@ -3,7 +3,7 @@
 import os
 import re
 from functools import lru_cache
-from typing import Any
+from typing import Any, cast
 
 import httpx
 import pandas as pd
@@ -21,7 +21,7 @@
 logger = logging.getLogger(__name__)
 
 
-class hashabledict(dict):
+class hashabledict(dict[str, Any]):
     """Hashable dict type used for caching."""
 
     def __hash__(self) -> int:
@@ -51,7 +51,7 @@ class JtiKwargs(TypedDict, total=False):
 class ManagerConfig(Config):
     template_pattern = re.compile(r"(\{env\:(\w+)\})")
 
-    def __init__(self, url: str, auth: BaseAuth, remote: bool = False) -> None:
+    def __init__(self, url: str, auth: BaseAuth | None, remote: bool = False) -> None:
         # TODO: Find the sweet-spot for `maxsize`
         # -> a trade-off between memory usage
         # and load on the management service
@@ -86,7 +86,7 @@ def _uncached_request(
         method: str,
         path: str,
         jti: str | None = None,
-    ) -> dict:
+    ) -> dict[str, Any]:
         del jti
         # `jti` is only used to affect `@lru_cache`
         # if the token id changes a new cache entry will be created
@@ -99,15 +99,16 @@ def _uncached_request(
         if res.status_code != 200:
             raise ManagerApiError(f"[{str(res.status_code)}] {res.text}")
         # TODO Can we really assume this type?
-        json: dict = res.json()
-        return json
+        return cast(dict[str, Any], res.json())
 
     def _request(
         self,
         method: str,
         path: str,
-        params: dict | None = None,
-        json: dict | list | tuple | None = None,
+        # Seems to be just that based on references
+        params: dict[str, int | None] | None = None,
+        # Seems to not be included with any references?
+        json: dict[str, Any] | list[Any] | tuple[Any] | None = None,
         **kwargs: Unpack[JtiKwargs],
     ) -> dict[str, Any]:
         if params is not None:
@@ -117,10 +118,7 @@ def _request(
             json = hashabledict(json) if isinstance(json, dict) else tuple(json)
 
         logger.debug(f"Trying cache: {method} {path} {params} {json}")
-        cached_request: dict[str, Any] = self._cached_request(
-            method, path, params=params, json=json, **kwargs
-        )
-        return cached_request
+        return self._cached_request(method, path, params=params, json=json, **kwargs)
 
     def fetch_platforms(self, **kwargs: Unpack[JtiKwargs]) -> list[ManagerPlatformInfo]:
         json = self._request("GET", "/ixmp4", params={"page_size": -1}, **kwargs)

ixmp4/conf/settings.py

@@ -82,10 +82,10 @@ def toml(self) -> TomlConfig:
         return self._toml  # type: ignore[return-value]
 
     @property
-    def default_auth(self) -> ManagerAuth | AnonymousAuth:
+    def default_auth(self) -> ManagerAuth | AnonymousAuth | None:
         if self._default_auth is None:
             self.get_auth()
-        return self._default_auth  # type: ignore[return-value]
+        return self._default_auth
 
     @property
     def manager(self) -> ManagerConfig:

ixmp4/core/decorators.py

@@ -1,6 +1,6 @@
 import functools
 from collections.abc import Callable
-from typing import Any
+from typing import Any, TypeVar
 
 import pandera as pa
 from pandera.errors import SchemaError as PanderaSchemaError
@@ -13,15 +13,17 @@
 
 from .exceptions import SchemaError
 
+T = TypeVar("T")
+
 
 def check_types(
-    func: Callable[[BaseRepository, DataFrame], None],
-) -> Callable[[BaseRepository, DataFrame], None]:
+    func: Callable[[BaseRepository, DataFrame[T]], None],
+) -> Callable[[BaseRepository, DataFrame[T]], None]:
     checked_func = pa.check_types(func)
 
     @functools.wraps(func)
     def wrapper(
-        *args: Unpack[tuple[BaseRepository, DataFrame]],
+        *args: Unpack[tuple[BaseRepository, DataFrame[T]]],
         skip_validation: bool = False,
         **kwargs: Any,
     ) -> None:

ixmp4/core/exceptions.py

@@ -13,9 +13,12 @@ class ProgrammingError(Exception):
 ExcMeta: type = type(Exception)
 
 
-class RemoteExceptionMeta(ExcMeta):
+# TODO can't find a similar open issues, should I open one?
+# For reference, this occurred after adding disallow_subclassing_any
+# NOTE mypy seems to say Exception has type Any, don't see how we could change that
+class RemoteExceptionMeta(ExcMeta):  # type: ignore[misc]
     def __new__(
-        cls, name: str, bases: tuple, namespace: dict, **kwargs: Any
+        cls, name: str, bases: tuple[object], namespace: dict[str, Any], **kwargs: Any
     ) -> type["IxmpError"]:
         http_error_name = namespace.get("http_error_name", None)
         if http_error_name is not None:
@@ -34,7 +37,7 @@ class IxmpError(Exception, metaclass=RemoteExceptionMeta):
     _message: str = ""
     http_status_code: int = 500
     http_error_name: ClassVar[str] = "ixmp_error"
-    kwargs: dict
+    kwargs: dict[str, Any]
 
     def __init__(
         self,
@@ -69,7 +72,7 @@ def message(self) -> str:
         return message
 
     @classmethod
-    def from_dict(cls, dict_: dict) -> Self:
+    def from_dict(cls, dict_: dict[str, Any]) -> Self:
         return cls(message=dict_["message"], **dict_["kwargs"])
 
 

ixmp4/core/iamc/data.py

@@ -1,4 +1,5 @@
-from typing import Optional
+from collections.abc import Iterable
+from typing import Optional, TypeVar
 
 import pandas as pd
 import pandera as pa
@@ -49,9 +50,10 @@ def convert_to_std_format(df: pd.DataFrame, join_runs: bool) -> pd.DataFrame:
         df.rename(columns={"step_year": "year"}, inplace=True)
         time_col = "year"
     else:
+        T = TypeVar("T", bool, float, int, str)
 
-        def map_step_column(df: pd.Series) -> pd.Series:
-            df["time"] = df[MAP_STEP_COLUMN[df.type]]
+        def map_step_column(df: pd.Series[T]) -> pd.Series[T]:
+            df["time"] = df[MAP_STEP_COLUMN[str(df.type)]]
             return df
 
         df = df.apply(map_step_column, axis=1)
@@ -137,9 +139,9 @@ def remove(
     def tabulate(
         self,
         *,
-        variable: dict | None = None,
-        region: dict | None = None,
-        unit: dict | None = None,
+        variable: dict[str, str | Iterable[str]] | None = None,
+        region: dict[str, str | Iterable[str]] | None = None,
+        unit: dict[str, str | Iterable[str]] | None = None,
         raw: bool = False,
     ) -> pd.DataFrame:
         df = self.backend.iamc.datapoints.tabulate(

ixmp4/core/iamc/variable.py

@@ -37,7 +37,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.iamc.variables.docs.delete(self.id)
         else:
@@ -72,9 +72,8 @@ def tabulate(self, name: str | None = None) -> pd.DataFrame:
         return self.backend.iamc.variables.tabulate(name=name)
 
     def _get_variable_id(self, variable: str) -> int | None:
-        if variable is None:
-            return None
-        elif isinstance(variable, str):
+        # TODO this check seems kind of redundant...
+        if isinstance(variable, str):
             obj = self.backend.iamc.variables.get(variable)
             return obj.id
         else:

ixmp4/core/model.py

@@ -37,7 +37,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.models.docs.delete(self.id)
         else:
@@ -75,9 +75,8 @@ def tabulate(self, name: str | None = None) -> pd.DataFrame:
         return self.backend.models.tabulate(name=name)
 
     def _get_model_id(self, model: str) -> int | None:
-        if model is None:
-            return None
-        elif isinstance(model, str):
+        # TODO this check seems kind of redundant...
+        if isinstance(model, str):
             obj = self.backend.models.get(model)
             return obj.id
         else:

ixmp4/core/optimization/equation.py

@@ -55,13 +55,13 @@ def remove_data(self) -> None:
         ).data
 
     @property
-    def levels(self) -> list:
-        levels: list = self._model.data.get("levels", [])
+    def levels(self) -> list[float]:
+        levels: list[float] = self._model.data.get("levels", [])
         return levels
 
     @property
-    def marginals(self) -> list:
-        marginals: list = self._model.data.get("marginals", [])
+    def marginals(self) -> list[float]:
+        marginals: list[float] = self._model.data.get("marginals", [])
         return marginals
 
     @property
@@ -88,7 +88,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.optimization.equations.docs.delete(self.id)
         else:

ixmp4/core/optimization/parameter.py

@@ -13,7 +13,7 @@
 from ixmp4.core.base import BaseFacade, BaseModelFacade
 from ixmp4.data.abstract import Docs as DocsModel
 from ixmp4.data.abstract import Parameter as ParameterModel
-from ixmp4.data.abstract import Run
+from ixmp4.data.abstract import Run, Unit
 from ixmp4.data.abstract.optimization import Column
 
 
@@ -48,13 +48,13 @@ def add(self, data: dict[str, Any] | pd.DataFrame) -> None:
         ).data
 
     @property
-    def values(self) -> list:
-        values: list = self._model.data.get("values", [])
+    def values(self) -> list[float]:
+        values: list[float] = self._model.data.get("values", [])
         return values
 
     @property
-    def units(self) -> list:
-        units: list = self._model.data.get("units", [])
+    def units(self) -> list[Unit]:
+        units: list[Unit] = self._model.data.get("units", [])
         return units
 
     @property
@@ -81,7 +81,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.optimization.parameters.docs.delete(self.id)
         else:

ixmp4/core/optimization/scalar.py

@@ -81,7 +81,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.optimization.scalars.docs.delete(self.id)
         else:

ixmp4/core/optimization/table.py

@@ -69,7 +69,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.optimization.tables.docs.delete(self.id)
         else:

ixmp4/core/optimization/variable.py

@@ -55,13 +55,13 @@ def remove_data(self) -> None:
         ).data
 
     @property
-    def levels(self) -> list:
-        levels: list = self._model.data.get("levels", [])
+    def levels(self) -> list[float]:
+        levels: list[float] = self._model.data.get("levels", [])
         return levels
 
     @property
-    def marginals(self) -> list:
-        marginals: list = self._model.data.get("marginals", [])
+    def marginals(self) -> list[float]:
+        marginals: list[float] = self._model.data.get("marginals", [])
         return marginals
 
     @property
@@ -92,7 +92,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.optimization.variables.docs.delete(self.id)
         else:

ixmp4/core/region.py

@@ -49,7 +49,7 @@ def docs(self) -> str | None:
             return None
 
     @docs.setter
-    def docs(self, description: str) -> None:
+    def docs(self, description: str | None) -> None:
         if description is None:
             self.backend.regions.docs.delete(self.id)
         else: