diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index a23eac1bd..203c8a6de 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -3,37 +3,29 @@ 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 ClanUrl: + 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})" + def __repr__(self) -> str: + return f"ClanUrl({self.value})" - @member - @dataclass - class LOCAL: - value: Path # The path field holds the local path + def is_local(self) -> bool: + return isinstance(self.value, Path) - def __str__(self) -> str: - return f"{self.value}" # The __str__ method returns a custom string representation - - def __repr__(self) -> str: - return f"ClanUrl.LOCAL({self.value})" + def is_remote(self) -> bool: + return isinstance(self.value, str) # Parameters defined here will be DELETED from the nested uri @@ -56,7 +48,6 @@ class MachineData: # Define the ClanURI class class ClanURI: _orig_uri: str - _nested_uri: str _components: urllib.parse.ParseResult url: ClanUrl _machines: list[MachineData] @@ -72,13 +63,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( @@ -113,9 +104,9 @@ class ClanURI: ) match comb: case ("file", "", path, "", "", _) | ("", "", path, "", "", _): # type: ignore - url = ClanUrl.LOCAL.value(Path(path).expanduser().resolve()) # type: ignore + url = ClanUrl(Path(path).expanduser().resolve()) case _: - url = ClanUrl.REMOTE.value(comps.geturl()) # type: ignore + url = ClanUrl(comps.geturl()) return url @@ -150,6 +141,18 @@ class ClanURI: def get_url(self) -> str: return str(self.url) + def to_json(self) -> dict[str, Any]: + return { + "_orig_uri": self._orig_uri, + "url": str(self.url), + "machines": [dataclasses.asdict(m) for m in self._machines], + } + + def from_json(self, data: dict[str, Any]) -> None: + self._orig_uri = data["_orig_uri"] + self.url = data["url"] + self._machines = [MachineData(**m) for m in data["machines"]] + @classmethod def from_str( cls, # noqa 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..7b572dc86 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 @@ -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.url.is_local(): + self._flake_path = Path(str(self.data.url)) + elif self.data.url.is_remote(): + self._flake_path = Path(nix_metadata(str(self.data.url))["path"]) + else: + raise ClanError(f"Unsupported flake url: {self.data.url}") assert self._flake_path is not None return self._flake_path 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..7b47cdaa8 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.url.is_local(): + self.machine = Machine( + name=self.data.flake.flake_attr, + flake=Path(str(uri.url)), + ) + if uri.url.is_remote(): + self.machine = Machine( + name=self.data.flake.flake_attr, + flake=str(uri.url), + ) yield self.machine self.machine = None