diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index a23eac1bd..9cb387cb8 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -3,37 +3,37 @@ import dataclasses import urllib.parse import urllib.request from dataclasses import dataclass -from enum import Enum, member from pathlib import Path from typing import Any from .errors import ClanError -# Define an enum with different members that have different values -class ClanUrl(Enum): - # Use the dataclass decorator to add fields and methods to the members - @member - @dataclass - class REMOTE: - value: str # The url field holds the HTTP URL +@dataclass +class FlakeId: + _value: str | Path - def __str__(self) -> str: - return f"{self.value}" # The __str__ method returns a custom string representation + def __str__(self) -> str: + return f"{self._value}" # The __str__ method returns a custom string representation - def __repr__(self) -> str: - return f"ClanUrl.REMOTE({self.value})" + @property + def path(self) -> Path: + assert isinstance(self._value, Path) + return self._value - @member - @dataclass - class LOCAL: - value: Path # The path field holds the local path + @property + def url(self) -> str: + assert isinstance(self._value, str) + return self._value - def __str__(self) -> str: - return f"{self.value}" # The __str__ method returns a custom string representation + def __repr__(self) -> str: + return f"ClanUrl({self._value})" - def __repr__(self) -> str: - return f"ClanUrl.LOCAL({self.value})" + def is_local(self) -> bool: + return isinstance(self._value, Path) + + def is_remote(self) -> bool: + return isinstance(self._value, str) # Parameters defined here will be DELETED from the nested uri @@ -45,20 +45,19 @@ class MachineParams: @dataclass class MachineData: - url: ClanUrl + flake_id: FlakeId name: str = "defaultVM" params: MachineParams = dataclasses.field(default_factory=MachineParams) def get_id(self) -> str: - return f"{self.url}#{self.name}" + return f"{self.flake_id}#{self.name}" # Define the ClanURI class class ClanURI: _orig_uri: str - _nested_uri: str _components: urllib.parse.ParseResult - url: ClanUrl + flake_id: FlakeId _machines: list[MachineData] # Initialize the class with a clan:// URI @@ -72,13 +71,13 @@ class ClanURI: # Check if the URI starts with clan:// # If it does, remove the clan:// prefix if uri.startswith("clan://"): - self._nested_uri = uri[7:] + nested_uri = uri[7:] else: raise ClanError(f"Invalid uri: expected clan://, got {uri}") # Parse the URI into components # url://netloc/path;parameters?query#fragment - self._components = urllib.parse.urlparse(self._nested_uri) + self._components = urllib.parse.urlparse(nested_uri) # Replace the query string in the components with the new query string clean_comps = self._components._replace( @@ -86,7 +85,7 @@ class ClanURI: ) # Parse the URL into a ClanUrl object - self.url = self._parse_url(clean_comps) + self.flake_id = self._parse_url(clean_comps) # Parse the fragment into a list of machine queries # Then parse every machine query into a MachineParameters object @@ -99,10 +98,10 @@ class ClanURI: # If there are no machine fragments, add a default machine if len(machine_frags) == 0: - default_machine = MachineData(url=self.url) + default_machine = MachineData(flake_id=self.flake_id) self._machines.append(default_machine) - def _parse_url(self, comps: urllib.parse.ParseResult) -> ClanUrl: + def _parse_url(self, comps: urllib.parse.ParseResult) -> FlakeId: comb = ( comps.scheme, comps.netloc, @@ -113,11 +112,11 @@ class ClanURI: ) match comb: case ("file", "", path, "", "", _) | ("", "", path, "", "", _): # type: ignore - url = ClanUrl.LOCAL.value(Path(path).expanduser().resolve()) # type: ignore + flake_id = FlakeId(Path(path).expanduser().resolve()) case _: - url = ClanUrl.REMOTE.value(comps.geturl()) # type: ignore + flake_id = FlakeId(comps.geturl()) - return url + return flake_id def _parse_machine_query(self, machine_frag: str) -> MachineData: comp = urllib.parse.urlparse(machine_frag) @@ -137,7 +136,7 @@ class ClanURI: # we need to make sure there are no conflicts del query[dfield.name] params = MachineParams(**machine_params) - machine = MachineData(url=self.url, name=machine_name, params=params) + machine = MachineData(flake_id=self.flake_id, name=machine_name, params=params) return machine @property @@ -148,7 +147,7 @@ class ClanURI: return self._orig_uri def get_url(self) -> str: - return str(self.url) + return str(self.flake_id) @classmethod def from_str( diff --git a/pkgs/clan-cli/clan_cli/history/add.py b/pkgs/clan-cli/clan_cli/history/add.py index 3b4b21ddc..f3b5ab7aa 100644 --- a/pkgs/clan-cli/clan_cli/history/add.py +++ b/pkgs/clan-cli/clan_cli/history/add.py @@ -17,13 +17,6 @@ from ..locked_open import read_history_file, write_history_file log = logging.getLogger(__name__) -class EnhancedJSONEncoder(json.JSONEncoder): - def default(self, o: Any) -> Any: - if dataclasses.is_dataclass(o): - return dataclasses.asdict(o) - return super().default(o) - - @dataclasses.dataclass class HistoryEntry: last_used: str diff --git a/pkgs/clan-cli/clan_cli/jsonrpc.py b/pkgs/clan-cli/clan_cli/jsonrpc.py new file mode 100644 index 000000000..0d779ee45 --- /dev/null +++ b/pkgs/clan-cli/clan_cli/jsonrpc.py @@ -0,0 +1,15 @@ +import dataclasses +import json +from typing import Any + + +class ClanJSONEncoder(json.JSONEncoder): + def default(self, o: Any) -> Any: + # Check if the object has a to_json method + if hasattr(o, "to_json") and callable(o.to_json): + return o.to_json() + # Check if the object is a dataclass + elif dataclasses.is_dataclass(o): + return dataclasses.asdict(o) + # Otherwise, use the default serialization + return super().default(o) diff --git a/pkgs/clan-cli/clan_cli/locked_open.py b/pkgs/clan-cli/clan_cli/locked_open.py index b8ea03b9f..a049d0ca3 100644 --- a/pkgs/clan-cli/clan_cli/locked_open.py +++ b/pkgs/clan-cli/clan_cli/locked_open.py @@ -1,4 +1,3 @@ -import dataclasses import fcntl import json from collections.abc import Generator @@ -6,16 +5,11 @@ from contextlib import contextmanager from pathlib import Path from typing import Any +from clan_cli.jsonrpc import ClanJSONEncoder + from .dirs import user_history_file -class EnhancedJSONEncoder(json.JSONEncoder): - def default(self, o: Any) -> Any: - if dataclasses.is_dataclass(o): - return dataclasses.asdict(o) - return super().default(o) - - @contextmanager def _locked_open(filename: str | Path, mode: str = "r") -> Generator: """ @@ -29,7 +23,7 @@ def _locked_open(filename: str | Path, mode: str = "r") -> Generator: def write_history_file(data: Any) -> None: with _locked_open(user_history_file(), "w+") as f: - f.write(json.dumps(data, cls=EnhancedJSONEncoder, indent=4)) + f.write(json.dumps(data, cls=ClanJSONEncoder, indent=4)) def read_history_file() -> list[dict]: diff --git a/pkgs/clan-cli/clan_cli/machines/machines.py b/pkgs/clan-cli/clan_cli/machines/machines.py index 629e256a3..72e4b9f99 100644 --- a/pkgs/clan-cli/clan_cli/machines/machines.py +++ b/pkgs/clan-cli/clan_cli/machines/machines.py @@ -6,7 +6,7 @@ from pathlib import Path from tempfile import NamedTemporaryFile from typing import Any -from clan_cli.clan_uri import ClanURI, ClanUrl, MachineData +from clan_cli.clan_uri import ClanURI, MachineData from clan_cli.dirs import vm_state_dir from qemu.qmp import QEMUMonitorProtocol @@ -66,7 +66,7 @@ class Machine: if machine is None: uri = ClanURI.from_str(str(flake), name) machine = uri.machine - self.flake: str | Path = machine.url.value + self.flake: str | Path = machine.flake_id._value self.name: str = machine.name self.data: MachineData = machine else: @@ -77,12 +77,12 @@ class Machine: self._flake_path: Path | None = None self._deployment_info: None | dict[str, str] = deployment_info - state_dir = vm_state_dir(flake_url=str(self.data.url), vm_name=self.data.name) + state_dir = vm_state_dir(flake_url=str(self.flake), vm_name=self.data.name) self.vm: QMPWrapper = QMPWrapper(state_dir) def __str__(self) -> str: - return f"Machine(name={self.data.name}, flake={self.data.url})" + return f"Machine(name={self.data.name}, flake={self.data.flake_id})" def __repr__(self) -> str: return str(self) @@ -139,11 +139,12 @@ class Machine: if self._flake_path: return self._flake_path - match self.data.url: - case ClanUrl.LOCAL.value(path): - self._flake_path = path - case ClanUrl.REMOTE.value(url): - self._flake_path = Path(nix_metadata(url)["path"]) + if self.data.flake_id.is_local(): + self._flake_path = self.data.flake_id.path + elif self.data.flake_id.is_remote(): + self._flake_path = Path(nix_metadata(self.data.flake_id.url)["path"]) + else: + raise ClanError(f"Unsupported flake url: {self.data.flake_id}") assert self._flake_path is not None return self._flake_path diff --git a/pkgs/clan-cli/tests/test_clan_uri.py b/pkgs/clan-cli/tests/test_clan_uri.py index f8c09cb7a..efddafa64 100644 --- a/pkgs/clan-cli/tests/test_clan_uri.py +++ b/pkgs/clan-cli/tests/test_clan_uri.py @@ -1,6 +1,6 @@ from pathlib import Path -from clan_cli.clan_uri import ClanURI, ClanUrl +from clan_cli.clan_uri import ClanURI def test_get_url() -> None: @@ -21,22 +21,13 @@ def test_get_url() -> None: def test_local_uri() -> None: # Create a ClanURI object from a local URI uri = ClanURI("clan://file:///home/user/Downloads") - match uri.url: - case ClanUrl.LOCAL.value(path): - assert path == Path("/home/user/Downloads") # type: ignore - case _: - assert False + assert uri.flake_id.path == Path("/home/user/Downloads") def test_is_remote() -> None: # Create a ClanURI object from a remote URI uri = ClanURI("clan://https://example.com") - - match uri.url: - case ClanUrl.REMOTE.value(url): - assert url == "https://example.com" # type: ignore - case _: - assert False + assert uri.flake_id.url == "https://example.com" def test_direct_local_path() -> None: @@ -56,12 +47,7 @@ def test_remote_with_clanparams() -> None: uri = ClanURI("clan://https://example.com") assert uri.machine.name == "defaultVM" - - match uri.url: - case ClanUrl.REMOTE.value(url): - assert url == "https://example.com" # type: ignore - case _: - assert False + assert uri.flake_id.url == "https://example.com" def test_remote_with_all_params() -> None: @@ -69,11 +55,7 @@ def test_remote_with_all_params() -> None: assert uri.machine.name == "myVM" assert uri._machines[1].name == "secondVM" assert uri._machines[1].params.dummy_opt == "1" - match uri.url: - case ClanUrl.REMOTE.value(url): - assert url == "https://example.com?password=12345" # type: ignore - case _: - assert False + assert uri.flake_id.url == "https://example.com?password=12345" def test_from_str_remote() -> None: @@ -82,11 +64,7 @@ def test_from_str_remote() -> None: assert uri.get_orig_uri() == "clan://https://example.com#myVM" assert uri.machine.name == "myVM" assert len(uri._machines) == 1 - match uri.url: - case ClanUrl.REMOTE.value(url): - assert url == "https://example.com" # type: ignore - case _: - assert False + assert uri.flake_id.url == "https://example.com" def test_from_str_local() -> None: @@ -95,11 +73,8 @@ def test_from_str_local() -> None: assert uri.get_orig_uri() == "clan://~/Projects/democlan#myVM" assert uri.machine.name == "myVM" assert len(uri._machines) == 1 - match uri.url: - case ClanUrl.LOCAL.value(path): - assert str(path).endswith("/Projects/democlan") # type: ignore - case _: - assert False + assert uri.flake_id.is_local() + assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore def test_from_str_local_no_machine() -> None: @@ -108,11 +83,8 @@ def test_from_str_local_no_machine() -> None: assert uri.get_orig_uri() == "clan://~/Projects/democlan" assert uri.machine.name == "defaultVM" assert len(uri._machines) == 1 - match uri.url: - case ClanUrl.LOCAL.value(path): - assert str(path).endswith("/Projects/democlan") # type: ignore - case _: - assert False + assert uri.flake_id.is_local() + assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore def test_from_str_local_no_machine2() -> None: @@ -121,8 +93,5 @@ def test_from_str_local_no_machine2() -> None: assert uri.get_orig_uri() == "clan://~/Projects/democlan#syncthing-peer1" assert uri.machine.name == "syncthing-peer1" assert len(uri._machines) == 1 - match uri.url: - case ClanUrl.LOCAL.value(path): - assert str(path).endswith("/Projects/democlan") # type: ignore - case _: - assert False + assert uri.flake_id.is_local() + assert str(uri.flake_id).endswith("/Projects/democlan") # type: ignore diff --git a/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py b/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py index 972257c8b..02a7ada31 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py +++ b/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, ClanUrl +from clan_cli.clan_uri import ClanURI from clan_cli.history.add import HistoryEntry from clan_cli.machines.machines import Machine @@ -115,17 +115,16 @@ class VMObject(GObject.Object): uri = ClanURI.from_str( url=self.data.flake.flake_url, machine_name=self.data.flake.flake_attr ) - match uri.url: - case ClanUrl.LOCAL.value(path): - self.machine = Machine( - name=self.data.flake.flake_attr, - flake=path, # type: ignore - ) - case ClanUrl.REMOTE.value(url): - self.machine = Machine( - name=self.data.flake.flake_attr, - flake=url, # type: ignore - ) + if uri.flake_id.is_local(): + self.machine = Machine( + name=self.data.flake.flake_attr, + flake=uri.flake_id.path, + ) + if uri.flake_id.is_remote(): + self.machine = Machine( + name=self.data.flake.flake_attr, + flake=uri.flake_id.url, + ) yield self.machine self.machine = None diff --git a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py index 50072c59a..71854844d 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py @@ -111,7 +111,7 @@ class ClanStore: del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr] def get_vm(self, uri: ClanURI) -> None | VMObject: - vm_store = self.clan_store.get(str(uri.url)) + vm_store = self.clan_store.get(str(uri.flake_id)) if vm_store is None: return None machine = vm_store.get(uri.machine.name, None)