clan-vm-manager: Fix regression

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

@@ -13,7 +13,7 @@ from typing import IO, ClassVar
 
 import gi
 from clan_cli import vms
-from clan_cli.clan_uri import ClanURI
+from clan_cli.clan_uri import ClanURI, FlakeId
 from clan_cli.dirs import vm_state_dir
 from clan_cli.history.add import HistoryEntry
 from clan_cli.machines.machines import Machine
@@ -51,7 +51,7 @@ class VMObject(GObject.Object):
 
         # Store the data from the history entry
         self.data: HistoryEntry = data
-
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         self.build_log_cb = build_log_cb
 
         # Create a process object to store the VM process
@@ -99,6 +99,7 @@ class VMObject(GObject.Object):
         return GLib.SOURCE_REMOVE
 
     def update(self, data: HistoryEntry) -> None:
+        assert isinstance(data.flake.flake_url, FlakeId)
         self.data = data
 
     def _on_vm_status_changed(self, source: "VMObject") -> None:
@@ -180,12 +181,14 @@ class VMObject(GObject.Object):
             return GLib.SOURCE_REMOVE
 
     def __start(self) -> None:
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         with self._create_machine() as machine:
             # Start building VM
             tstart = datetime.now()
             log.info(f"Building VM {self.get_id()}")
             log_dir = Path(str(self.log_dir.name))
 
+            assert isinstance(self.data.flake.vm.flake_url, FlakeId)
             # Start the build process
             self.build_process = spawn(
                 on_except=None,
@@ -272,6 +275,7 @@ class VMObject(GObject.Object):
             self.emit("vm_status_changed", self)
             return
         log.debug(f"VM state dir {self.log_dir.name}")
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         self._start_thread = threading.Thread(target=self.__start)
         self._start_thread.start()
 

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

@@ -4,7 +4,7 @@ from pathlib import Path
 from typing import Any, ClassVar
 
 import gi
-from clan_cli.clan_uri import ClanURI
+from clan_cli.clan_uri import ClanURI, FlakeId
 from clan_cli.history.add import HistoryEntry
 from clan_cli.machines.machines import Machine
 
@@ -34,7 +34,7 @@ class Emitter(GObject.GObject):
 
 class ClanStore:
     _instance: "None | ClanStore" = None
-    _clan_store: GKVStore[str, VMStore]
+    _clan_store: GKVStore[FlakeId, VMStore]
 
     _emitter: Emitter
 
@@ -65,6 +65,7 @@ class ClanStore:
 
     def set_logging_vm(self, ident: str) -> VMObject | None:
         vm = self.get_vm(ClanURI(f"clan://{ident}"))
+
         if vm is not None:
             self._logging_vm = vm
 
@@ -92,7 +93,7 @@ class ClanStore:
         self.clan_store.register_on_change(on_clanstore_change)
 
     @property
-    def clan_store(self) -> GKVStore[str, VMStore]:
+    def clan_store(self) -> GKVStore[FlakeId, VMStore]:
         return self._clan_store
 
     def create_vm_task(self, vm: HistoryEntry) -> bool:
@@ -109,8 +110,12 @@ class ClanStore:
         def log_details(gfile: Gio.File) -> None:
             self.log_details(vm, gfile)
 
+        assert isinstance(entry.flake.flake_url, FlakeId)
+
         vm = VMObject(icon=icon, data=entry, build_log_cb=log_details)
+        assert isinstance(vm.data.flake.flake_url, FlakeId)
         self.push(vm)
+        assert isinstance(vm.data.flake.flake_url, FlakeId)
 
     def log_details(self, vm: VMObject, gfile: Gio.File) -> None:
         views = ViewStack.use().view
@@ -136,7 +141,7 @@ class ClanStore:
         # we cannot check this type, python is not smart enough
 
     def push(self, vm: VMObject) -> None:
-        url = str(vm.data.flake.flake_url)
+        url = vm.data.flake.flake_url
 
         # Only write to the store if the Clan is not already in it
         # Every write to the KVStore rerenders bound widgets to the clan_store
@@ -160,11 +165,12 @@ class ClanStore:
                 vm_store.append(vm)
 
     def remove(self, vm: VMObject) -> None:
-        del self.clan_store[str(vm.data.flake.flake_url)][vm.data.flake.flake_attr]
+        del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
 
     def get_vm(self, uri: ClanURI) -> None | VMObject:
         machine = Machine(uri.machine_name, uri.flake)
-        vm_store = self.clan_store.get(str(machine.flake))
+        vm_store = self.clan_store.get(machine.flake)
+
         if vm_store is None:
             return None
         vm = vm_store.get(str(machine.name), None)

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

@@ -103,7 +103,7 @@ class ClanList(Gtk.Box):
 
         grp = Adw.PreferencesGroup()
         grp.set_title(vm.data.flake.clan_name)
-        grp.set_description(vm.data.flake.flake_url)
+        grp.set_description(str(vm.data.flake.flake_url))
 
         add_action = Gio.SimpleAction.new("add", GLib.VariantType.new("s"))
         add_action.connect("activate", self.on_add)

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

@@ -2,6 +2,7 @@ import logging
 import threading
 
 import gi
+from clan_cli.clan_uri import FlakeId
 from clan_cli.history.list import list_history
 
 from clan_vm_manager.components.interfaces import ClanConfig
@@ -73,6 +74,7 @@ class MainWindow(Adw.ApplicationWindow):
         # Execute `clan flakes add <path>` to democlan for this to work
         # TODO: Make list_history a generator function
         for entry in list_history():
+            assert isinstance(entry.flake.flake_url, FlakeId)
             GLib.idle_add(ClanStore.use().create_vm_task, entry)
 
         GLib.idle_add(self._set_clan_store_ready)