Enable professional-grade mypy

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.12.0
+  rev: v1.13.0
   hooks:
   - id: mypy
     entry: bash -c "poetry run mypy ."

README.md

@@ -5,6 +5,7 @@ Copyright (c) 2023-2024 IIASA - Energy, Climate, and Environment Program (ECE)
 [![license: MIT](https://img.shields.io/badge/license-MIT-brightgreen.svg)](https://github.com/iiasa/ixmp4/blob/main/LICENSE)
 [![python](https://img.shields.io/badge/python-3.10_|_3.11_|_3.12_|_3.13-blue?logo=python&logoColor=white)](https://github.com/iiasa/ixmp4)
 [![Code style: ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/charliermarsh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
+[![Checked with mypy](https://www.mypy-lang.org/static/mypy_badge.svg)](https://mypy-lang.org/)
 
 ## Overview
 

doc/source/conf.py

@@ -31,7 +31,7 @@
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
 # This pattern also affects html_static_path and html_extra_path.
-exclude_patterns = []
+exclude_patterns: list[str] = []
 
 
 # -- Options for HTML output -------------------------------------------------

ixmp4/cli/__init__.py

@@ -42,7 +42,7 @@
         prompt=True,
         hide_input=True,
     ),
-):
+) -> None:
     try:
         auth = ManagerAuth(username, password, str(settings.manager_url))
         user = auth.get_user()
@@ -62,7 +62,7 @@
 
 
 @app.command()
-def logout():
+def logout() -> None:
     if typer.confirm(
         "Are you sure you want to log out and delete locally saved credentials?"
     ):
@@ -80,7 +80,7 @@
         with_backend: Optional[bool] = False,
         with_benchmarks: Optional[bool] = False,
         dry: Optional[bool] = False,
-    ):
+    ) -> None:
         opts = [
             "--cov-report",
             "xml:.coverage.xml",

ixmp4/cli/platforms.py

@@ -1,7 +1,8 @@
 import re
+from collections.abc import Generator, Iterator
 from itertools import cycle
 from pathlib import Path
-from typing import Generator, Optional
+from typing import Any, Optional, TypeVar
 
 import typer
 from rich.progress import Progress, track
@@ -21,15 +22,15 @@
 app = typer.Typer()
 
 
-def validate_name(name: str):
+def validate_name(name: str) -> str:
     match = re.match(r"^[\w\-_]*$", name)
     if match is None:
         raise typer.BadParameter("Platform name must be slug-like.")
     else:
         return name
 
 
-def validate_dsn(dsn: str | None):
+def validate_dsn(dsn: str | None) -> str | None:
     if dsn is None:
         return None
     match = re.match(r"^(sqlite|postgresql\+psycopg|https|http)(\:\/\/)", dsn)
@@ -41,7 +42,7 @@
         return dsn
 
 
-def prompt_sqlite_dsn(name: str):
+def prompt_sqlite_dsn(name: str) -> str:
     path = sqlite.get_database_path(name)
     dsn = sqlite.get_dsn(path)
     if path.exists():
@@ -75,7 +76,7 @@
         help="Data source name. Can be a http(s) URL or a database connection string.",
         callback=validate_dsn,
     ),
-):
+) -> None:
     try:
         settings.toml.get_platform(name)
         raise typer.BadParameter(
@@ -95,11 +96,11 @@
     utils.good("\nPlatform added successfully.")
 
 
-def prompt_sqlite_removal(dsn: str):
+def prompt_sqlite_removal(dsn: str) -> None:
     path = Path(dsn.replace("sqlite://", ""))
     path_str = typer.style(path, fg=typer.colors.CYAN)
     if typer.confirm(
-        "Do you want to remove the associated database file at " f"{path_str} as well?"  # type: ignore
+        "Do you want to remove the associated database file at " f"{path_str} as well?"
     ):
         path.unlink()
         utils.echo("\nDatabase file deleted.")
@@ -112,7 +113,7 @@
     name: str = typer.Argument(
         ..., help="The string identifier of the platform to remove."
     ),
-):
+) -> None:
     try:
         platform = settings.toml.get_platform(name)
     except PlatformNotFound:
@@ -127,7 +128,7 @@
         settings.toml.remove_platform(name)
 
 
-def tabulate_toml_platforms(platforms: list[TomlPlatformInfo]):
+def tabulate_toml_platforms(platforms: list[TomlPlatformInfo]) -> None:
     toml_path_str = typer.style(settings.toml.path, fg=typer.colors.CYAN)
     utils.echo(f"\nPlatforms registered in '{toml_path_str}'")
     if len(platforms):
@@ -140,7 +141,7 @@
 
 def tabulate_manager_platforms(
     platforms: list[ManagerPlatformInfo],
-):
+) -> None:
     manager_url_str = typer.style(settings.manager.url, fg=typer.colors.CYAN)
     utils.echo(f"\nPlatforms accessible via '{manager_url_str}'")
     utils.echo("\nName".ljust(21) + "Access".ljust(10) + "Notice")
@@ -154,7 +155,7 @@
 
 
 @app.command("list", help="Lists all registered platforms.")
-def list_():
+def list_() -> None:
     tabulate_toml_platforms(settings.toml.list_platforms())
     if settings.manager is not None:
         tabulate_manager_platforms(settings.manager.list_platforms())
@@ -166,7 +167,8 @@
         "revision."
     )
 )
-def upgrade():
+def upgrade() -> None:
+    platform_list: list[ManagerPlatformInfo] | list[TomlPlatformInfo]
     if settings.managed:
         utils.echo(
             f"Establishing self-signed admin connection to '{settings.manager_url}'."
@@ -232,7 +234,7 @@
     num_datapoints: int = typer.Option(
         30_000, "--datapoints", help="Number of mock datapoints to generate."
     ),
-):
+) -> None:
     try:
         platform = Platform(platform_name)
     except PlatformNotFound:
@@ -270,7 +272,12 @@
         utils.good("Done!")
 
 
-def create_cycle(generator: Generator, name: str, total: int):
+T = TypeVar("T")
+
+
+def create_cycle(
+    generator: Generator[T, Any, None], name: str, total: int
+) -> Iterator[T]:
     return cycle(
         [
             m
@@ -283,7 +290,7 @@
     )
 
 
-def generate_data(generator: MockDataGenerator):
+def generate_data(generator: MockDataGenerator) -> None:
     model_names = create_cycle(
         generator.yield_model_names(), "Model", generator.num_models
     )
@@ -301,7 +308,7 @@
             progress.advance(task, len(df))
 
 
-def _shorten(value: str, length: int):
+def _shorten(value: str, length: int) -> str:
     """Shorten and adjust a string to a given length adding `...` if necessary"""
     if len(value) > length - 4:
         value = value[: length - 4] + "..."

ixmp4/cli/server.py

@@ -2,7 +2,7 @@
 from typing import Optional
 
 import typer
-import uvicorn  # type: ignore[import]
+import uvicorn
 from fastapi.openapi.utils import get_openapi
 
 from ixmp4.conf import settings
@@ -33,7 +33,9 @@
 
 
 @app.command()
-def dump_schema(output_file: Optional[typer.FileTextWrite] = typer.Option(None, "-o")):
+def dump_schema(
+    output_file: Optional[typer.FileTextWrite] = typer.Option(None, "-o"),
+) -> None:
     schema = get_openapi(
         title=v1.title,
         version=v1.version,

ixmp4/conf/__init__.py

@@ -3,6 +3,4 @@
 from ixmp4.conf.settings import Settings
 
 load_dotenv()
-# strict typechecking fails due to a bug
-# https://docs.pydantic.dev/visual_studio_code/#adding-a-default-with-field
-settings = Settings()  # type: ignore
+settings = Settings()

ixmp4/conf/auth.py

@@ -1,5 +1,7 @@
 import logging
+from collections.abc import Generator
 from datetime import datetime, timedelta
+from typing import Any, cast
 from uuid import uuid4
 
 import httpx
@@ -13,10 +15,11 @@
 
 
 class BaseAuth(object):
-    def __call__(self, *args, **kwargs):
+    # This should never be called
+    def __call__(self, *args: Any, **kwargs: Any) -> httpx.Request:
         raise NotImplementedError
 
-    def auth_flow(self, request):
+    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, Any, None]:
         yield self(request)
 
     def get_user(self) -> User:
@@ -39,7 +42,7 @@
         )
         self.token = self.get_local_jwt()
 
-    def __call__(self, r):
+    def __call__(self, r: httpx.Request) -> httpx.Request:
         try:
             jwt.decode(self.token, self.secret, algorithms=["HS256"])
         except (jwt.InvalidTokenError, jwt.ExpiredSignatureError):
@@ -48,7 +51,7 @@
         r.headers["Authorization"] = "Bearer " + self.token
         return r
 
-    def get_local_jwt(self):
+    def get_local_jwt(self) -> str:
         self.jti = uuid4().hex
         return jwt.encode(
             {
@@ -63,7 +66,7 @@
             algorithm="HS256",
         )
 
-    def get_expiration_timestamp(self):
+    def get_expiration_timestamp(self) -> int:
         return int((datetime.now() + timedelta(minutes=15)).timestamp())
 
     def get_user(self) -> User:
@@ -72,11 +75,11 @@
 
 
 class AnonymousAuth(BaseAuth, httpx.Auth):
-    def __init__(self):
+    def __init__(self) -> None:
         self.user = anonymous_user
         logger.info("Connecting to service anonymously and without credentials.")
 
-    def __call__(self, r):
+    def __call__(self, r: httpx.Request) -> httpx.Request:
         return r
 
     def get_user(self) -> User:
@@ -97,7 +100,7 @@
         self.password = password
         self.obtain_jwt()
 
-    def __call__(self, r):
+    def __call__(self, r: httpx.Request) -> httpx.Request:
         try:
             jwt.decode(
                 self.access_token,
@@ -109,7 +112,7 @@
         r.headers["Authorization"] = "Bearer " + self.access_token
         return r
 
-    def obtain_jwt(self):
+    def obtain_jwt(self) -> None:
         res = self.client.post(
             "/token/obtain/",
             json={
@@ -133,7 +136,7 @@
         self.set_user(self.access_token)
         self.refresh_token = json["refresh"]
 
-    def refresh_or_reobtain_jwt(self):
+    def refresh_or_reobtain_jwt(self) -> None:
         try:
             jwt.decode(
                 self.refresh_token,
@@ -143,7 +146,7 @@
         except jwt.ExpiredSignatureError:
             self.obtain_jwt()
 
-    def refresh_jwt(self):
+    def refresh_jwt(self) -> None:
         res = self.client.post(
             "/token/refresh/",
             json={
@@ -157,13 +160,16 @@
         self.access_token = res.json()["access"]
         self.set_user(self.access_token)
 
-    def decode_token(self, token: str):
-        return jwt.decode(
-            token,
-            options={"verify_signature": False, "verify_exp": False},
+    def decode_token(self, token: str) -> dict[str, Any]:
+        return cast(
+            dict[str, Any],
+            jwt.decode(
+                token,
+                options={"verify_signature": False, "verify_exp": False},
+            ),
         )
 
-    def set_user(self, token: str):
+    def set_user(self, token: str) -> None:
         token_dict = self.decode_token(token)
         user_dict = token_dict["user"]
         self.user = User(**user_dict, jti=token_dict.get("jti"))

ixmp4/conf/credentials.py

@@ -5,31 +5,31 @@
 
 
 class Credentials(object):
-    credentials: dict
+    credentials: dict[str, dict[str, str]]
 
     def __init__(self, toml_file: Path) -> None:
         self.path = toml_file
         self.load()
 
-    def load(self):
+    def load(self) -> None:
         self.credentials = toml.load(self.path)
 
-    def dump(self):
+    def dump(self) -> None:
         f = self.path.open("w+")
         toml.dump(self.credentials, f)
 
     def get(self, key: str) -> tuple[str, str]:
         c = self.credentials[key]
         return (c["username"], c["password"])
 
-    def set(self, key: str, username: str, password: str):
+    def set(self, key: str, username: str, password: str) -> None:
         self.credentials[key] = {
             "username": username,
             "password": password,
         }
         self.dump()
 
-    def clear(self, key: str):
+    def clear(self, key: str) -> None:
         with suppress(KeyError):
             del self.credentials[key]
         self.dump()