clan-vm-manager: Add library for mypy pygobject types

pkgs/clan-vm-manager/clan_vm_manager/app.py

@@ -33,10 +33,8 @@ class MainApplication(Adw.Application):
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(
-            *args,
             application_id="org.clan.vm-manager",
             flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE,
-            **kwargs,
         )
 
         self.add_main_option(
@@ -48,7 +46,7 @@ class MainApplication(Adw.Application):
             None,
         )
 
-        self.window: Adw.ApplicationWindow | None = None
+        self.window: "MainWindow" | None = None
         self.connect("activate", self.on_activate)
         self.connect("shutdown", self.on_shutdown)
 
@@ -113,8 +111,10 @@ class MainApplication(Adw.Application):
         log.debug(f"Style css path: {resource_path}")
         css_provider = Gtk.CssProvider()
         css_provider.load_from_path(str(resource_path))
+        display = Gdk.Display.get_default()
+        assert display is not None
         Gtk.StyleContext.add_provider_for_display(
-            Gdk.Display.get_default(),
+            display,
             css_provider,
             Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
         )

pkgs/clan-vm-manager/clan_vm_manager/components/gkvstore.py

@@ -134,8 +134,8 @@ class GKVStore(GObject.GObject, Gio.ListModel, Generic[K, V]):
     def do_get_item(self, position: int) -> V | None:
         return self.get_item(position)
 
-    def get_item_type(self) -> GObject.GType:
-        return self.gtype.__gtype__
+    def get_item_type(self) -> Any:
+        return self.gtype.__gtype__  # type: ignore[attr-defined]
 
     def do_get_item_type(self) -> GObject.GType:
         return self.get_item_type()
@@ -187,10 +187,10 @@ class GKVStore(GObject.GObject, Gio.ListModel, Generic[K, V]):
         return len(self._items)
 
     # O(1) operation
-    def __getitem__(self, key: K) -> V:
+    def __getitem__(self, key: K) -> V:  # type: ignore[override]
         return self._items[key]
 
-    def __contains__(self, key: K) -> bool:
+    def __contains__(self, key: K) -> bool:  # type: ignore[override]
         return key in self._items
 
     def __str__(self) -> str:

pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py

@@ -24,6 +24,9 @@ import sys
 from collections.abc import Callable
 from typing import Any, ClassVar
 
+import gi
+
+gi.require_version("Gtk", "4.0")
 from gi.repository import GdkPixbuf, Gio, GLib, Gtk
 
 
@@ -168,7 +171,7 @@ class ImplUnavailableError(Exception):
 
 
 class BaseImplementation:
-    def __init__(self, application: Gtk.Application) -> None:
+    def __init__(self, application: Any) -> None:
         self.application = application
         self.menu_items: dict[int, Any] = {}
         self.menu_item_id: int = 1
@@ -1090,8 +1093,8 @@ class Win32Implementation(BaseImplementation):
 
 
 class TrayIcon:
-    def __init__(self, application: Gtk.Application) -> None:
-        self.application: Gtk.Application = application
+    def __init__(self, application: Gio.Application) -> None:
+        self.application: Gio.Application = application
         self.available: bool = True
         self.implementation: Any = None
 

pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py

@@ -32,13 +32,6 @@ class VMObject(GObject.Object):
         "vm_status_changed": (GObject.SignalFlags.RUN_FIRST, None, [])
     }
 
-    def _vm_status_changed_task(self) -> bool:
-        self.emit("vm_status_changed")
-        return GLib.SOURCE_REMOVE
-
-    def update(self, data: HistoryEntry) -> None:
-        self.data = data
-
     def __init__(
         self,
         icon: Path,
@@ -47,16 +40,20 @@ class VMObject(GObject.Object):
         super().__init__()
 
         # Store the data from the history entry
-        self.data = data
+        self.data: HistoryEntry = data
 
         # Create a process object to store the VM process
-        self.vm_process = MPProcess("vm_dummy", mp.Process(), Path("./dummy"))
-        self.build_process = MPProcess("build_dummy", mp.Process(), Path("./dummy"))
+        self.vm_process: MPProcess = MPProcess(
+            "vm_dummy", mp.Process(), Path("./dummy")
+        )
+        self.build_process: MPProcess = MPProcess(
+            "build_dummy", mp.Process(), Path("./dummy")
+        )
         self._start_thread: threading.Thread = threading.Thread()
         self.machine: Machine | None = None
 
         # Watcher to stop the VM
-        self.KILL_TIMEOUT = 20  # seconds
+        self.KILL_TIMEOUT: int = 20  # seconds
         self._stop_thread: threading.Thread = threading.Thread()
 
         # Build progress bar vars
@@ -66,7 +63,7 @@ class VMObject(GObject.Object):
         self.prog_bar_id: int = 0
 
         # Create a temporary directory to store the logs
-        self.log_dir = tempfile.TemporaryDirectory(
+        self.log_dir: tempfile.TemporaryDirectory = tempfile.TemporaryDirectory(
             prefix="clan_vm-", suffix=f"-{self.data.flake.flake_attr}"
         )
         self._logs_id: int = 0
@@ -75,14 +72,21 @@ class VMObject(GObject.Object):
         # To be able to set the switch state programmatically
         # we need to store the handler id returned by the connect method
         # and block the signal while we change the state. This is cursed.
-        self.switch = Gtk.Switch()
+        self.switch: Gtk.Switch = Gtk.Switch()
         self.switch_handler_id: int = self.switch.connect(
             "notify::active", self._on_switch_toggle
         )
         self.connect("vm_status_changed", self._on_vm_status_changed)
 
         # Make sure the VM is killed when the reference to this object is dropped
-        self._finalizer = weakref.finalize(self, self._kill_ref_drop)
+        self._finalizer: weakref.finalize = weakref.finalize(self, self._kill_ref_drop)
+
+    def _vm_status_changed_task(self) -> bool:
+        self.emit("vm_status_changed")
+        return GLib.SOURCE_REMOVE
+
+    def update(self, data: HistoryEntry) -> None:
+        self.data = data
 
     def _on_vm_status_changed(self, source: "VMObject") -> None:
         self.switch.set_state(self.is_running() and not self.is_building())
@@ -93,9 +97,8 @@ class VMObject(GObject.Object):
         exit_build = self.build_process.proc.exitcode
         exitc = exit_vm or exit_build
         if not self.is_running() and exitc != 0:
-            self.switch.handler_block(self.switch_handler_id)
-            self.switch.set_active(False)
-            self.switch.handler_unblock(self.switch_handler_id)
+            with self.switch.handler_block(self.switch_handler_id):
+                self.switch.set_active(False)
             log.error(f"VM exited with error. Exitcode: {exitc}")
 
     def _on_switch_toggle(self, switch: Gtk.Switch, user_state: bool) -> None:

pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py

@@ -1,7 +1,7 @@
 import logging
 import threading
 from collections.abc import Callable
-from typing import Any, ClassVar
+from typing import Any, ClassVar, cast
 
 import gi
 from clan_cli.clan_uri import ClanURI
@@ -31,8 +31,8 @@ class JoinValue(GObject.Object):
 
     def __init__(self, url: ClanURI) -> None:
         super().__init__()
-        self.url = url
-        self.entry = None
+        self.url: ClanURI = url
+        self.entry: HistoryEntry | None = None
 
     def __join(self) -> None:
         new_entry = add_history(self.url)
@@ -84,7 +84,7 @@ class JoinList:
 
         value = JoinValue(uri)
         if value.url.machine.get_id() in [
-            item.url.machine.get_id() for item in self.list_store
+            cast(JoinValue, item).url.machine.get_id() for item in self.list_store
         ]:
             log.info(f"Join request already exists: {value.url}. Ignoring.")
             return

pkgs/clan-vm-manager/clan_vm_manager/views/details.py

@@ -1,16 +1,19 @@
 import os
 from collections.abc import Callable
 from functools import partial
-from typing import Any, Literal
+from typing import Any, Literal, TypeVar
 
 import gi
 
 gi.require_version("Adw", "1")
 from gi.repository import Adw, Gio, GObject, Gtk
 
+# Define a TypeVar that is bound to GObject.Object
+ListItem = TypeVar("ListItem", bound=GObject.Object)
+
 
 def create_details_list(
-    model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, GObject], Gtk.Widget]
+    model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, ListItem], Gtk.Widget]
 ) -> Gtk.ListBox:
     boxed_list = Gtk.ListBox()
     boxed_list.set_selection_mode(Gtk.SelectionMode.NONE)
@@ -49,7 +52,10 @@ class Details(Gtk.Box):
     def render_entry_row(
         self, boxed_list: Gtk.ListBox, item: PreferencesValue
     ) -> Gtk.Widget:
-        row = Adw.SpinRow.new_with_range(0, os.cpu_count(), 1)
+        cores: int | None = os.cpu_count()
+        fcores = float(cores) if cores else 1.0
+
+        row = Adw.SpinRow.new_with_range(0, fcores, 1)
         row.set_value(item.data)
 
         return row

pkgs/clan-vm-manager/clan_vm_manager/views/list.py

@@ -1,7 +1,7 @@
 import logging
 from collections.abc import Callable
 from functools import partial
-from typing import Any
+from typing import Any, TypeVar
 
 import gi
 from clan_cli import history
@@ -17,9 +17,13 @@ from gi.repository import Adw, Gdk, Gio, GLib, GObject, Gtk
 
 log = logging.getLogger(__name__)
 
+ListItem = TypeVar("ListItem", bound=GObject.Object)
+CustomStore = TypeVar("CustomStore", bound=Gio.ListModel)
+
 
 def create_boxed_list(
-    model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, GObject], Gtk.Widget]
+    model: CustomStore,
+    render_row: Callable[[Gtk.ListBox, ListItem], Gtk.Widget],
 ) -> Gtk.ListBox:
     boxed_list = Gtk.ListBox()
     boxed_list.set_selection_mode(Gtk.SelectionMode.NONE)
@@ -47,8 +51,9 @@ class ClanList(Gtk.Box):
     def __init__(self, config: ClanConfig) -> None:
         super().__init__(orientation=Gtk.Orientation.VERTICAL)
 
-        self.app = Gio.Application.get_default()
-        self.app.connect("join_request", self.on_join_request)
+        app = Gio.Application.get_default()
+        assert app is not None
+        app.connect("join_request", self.on_join_request)
 
         self.log_label: Gtk.Label = Gtk.Label()
         self.__init_machines = history.add.list_history()
@@ -78,6 +83,7 @@ class ClanList(Gtk.Box):
         add_action = Gio.SimpleAction.new("add", GLib.VariantType.new("s"))
         add_action.connect("activate", self.on_add)
         app = Gio.Application.get_default()
+        assert app is not None
         app.add_action(add_action)
 
         menu_model = Gio.Menu()
@@ -158,6 +164,7 @@ class ClanList(Gtk.Box):
         open_action = Gio.SimpleAction.new("edit", GLib.VariantType.new("s"))
         open_action.connect("activate", self.on_edit)
         app = Gio.Application.get_default()
+        assert app is not None
         app.add_action(open_action)
         menu_model = Gio.Menu()
         menu_model.append("Edit", f"app.edit::{vm.get_id()}")
@@ -199,6 +206,7 @@ class ClanList(Gtk.Box):
         # Can't do this here because clan store is empty at this point
         if vm is not None:
             sub = row.get_subtitle()
+            assert sub is not None
             row.set_subtitle(
                 sub + "\nClan already exists. Joining again will update it"
             )

pkgs/clan-vm-manager/clan_vm_manager/windows/main_window.py

@@ -32,6 +32,7 @@ class MainWindow(Adw.ApplicationWindow):
         view.add_top_bar(header)
 
         app = Gio.Application.get_default()
+        assert app is not None
         self.tray_icon: TrayIcon = TrayIcon(app)
 
         # Initialize all ClanStore

pkgs/clan-vm-manager/default.nix

@@ -6,6 +6,7 @@
 , wrapGAppsHook
 , gtk4
 , gnome
+, pygobject-stubs
 , gobject-introspection
 , clan-cli
 , makeDesktopItem
@@ -41,7 +42,9 @@ python3.pkgs.buildPythonApplication {
   ];
 
   buildInputs = [ gtk4 libadwaita gnome.adwaita-icon-theme ];
-  propagatedBuildInputs = [ pygobject3 clan-cli ];
+
+  # We need to propagate the build inputs to nix fmt / treefmt
+  propagatedBuildInputs = [ pygobject3 clan-cli pygobject-stubs ];
 
   # also re-expose dependencies so we test them in CI
   passthru = {

pkgs/clan-vm-manager/pyproject.toml

@@ -22,9 +22,9 @@ disallow_untyped_calls = true
 disallow_untyped_defs = true
 no_implicit_optional = true
 
-[[tool.mypy.overrides]]
-module = "gi.*"
-ignore_missing_imports = true
+# [[tool.mypy.overrides]]
+# module = "gi.*"
+# ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
 module = "clan_cli.*"

pkgs/clan-vm-manager/shell.nix

@@ -24,7 +24,7 @@ mkShell (
       python3Packages.ipdb
       gtk4.dev
       libadwaita.devdoc # has the demo called 'adwaita-1-demo'
-    ] ++ clan-vm-manager.nativeBuildInputs;
+    ] ++ clan-vm-manager.nativeBuildInputs ++ clan-vm-manager.propagatedBuildInputs;
 
     PYTHONBREAKPOINT = "ipdb.set_trace";