forked from clan/clan-core
clan_cli: Simplify ClanURI
This commit is contained in:
parent
93c868a3b7
commit
442e5b45ba
@ -19,7 +19,10 @@ class ClanUrl(Enum):
|
||||
url: str # The url field holds the HTTP URL
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"REMOTE({self.url})" # The __str__ method returns a custom string representation
|
||||
return f"{self.url}" # The __str__ method returns a custom string representation
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ClanUrl.REMOTE({self.url})"
|
||||
|
||||
@member
|
||||
@dataclass
|
||||
@ -27,7 +30,10 @@ class ClanUrl(Enum):
|
||||
path: Path # The path field holds the local path
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"LOCAL({self.path})" # The __str__ method returns a custom string representation
|
||||
return f"{self.path}" # The __str__ method returns a custom string representation
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ClanUrl.LOCAL({self.path})"
|
||||
|
||||
|
||||
# Parameters defined here will be DELETED from the nested uri
|
||||
@ -39,9 +45,13 @@ class MachineParams:
|
||||
|
||||
@dataclass
|
||||
class MachineData:
|
||||
url: ClanUrl
|
||||
name: str = "defaultVM"
|
||||
params: MachineParams = dataclasses.field(default_factory=MachineParams)
|
||||
|
||||
def get_id(self) -> str:
|
||||
return f"{self.url}#{self.name}"
|
||||
|
||||
|
||||
# Define the ClanURI class
|
||||
class ClanURI:
|
||||
@ -49,11 +59,11 @@ class ClanURI:
|
||||
_nested_uri: str
|
||||
_components: urllib.parse.ParseResult
|
||||
url: ClanUrl
|
||||
machines: list[MachineData]
|
||||
_machines: list[MachineData]
|
||||
|
||||
# Initialize the class with a clan:// URI
|
||||
def __init__(self, uri: str) -> None:
|
||||
self.machines = []
|
||||
self._machines = []
|
||||
|
||||
# users might copy whitespace along with the uri
|
||||
uri = uri.strip()
|
||||
@ -85,11 +95,12 @@ class ClanURI:
|
||||
)
|
||||
for machine_frag in machine_frags:
|
||||
machine = self._parse_machine_query(machine_frag)
|
||||
self.machines.append(machine)
|
||||
self._machines.append(machine)
|
||||
|
||||
# If there are no machine fragments, add a default machine
|
||||
if len(machine_frags) == 0:
|
||||
self.machines.append(MachineData())
|
||||
default_machine = MachineData(url=self.url)
|
||||
self._machines.append(default_machine)
|
||||
|
||||
def _parse_url(self, comps: urllib.parse.ParseResult) -> ClanUrl:
|
||||
comb = (
|
||||
@ -126,20 +137,35 @@ class ClanURI:
|
||||
# we need to make sure there are no conflicts
|
||||
del query[dfield.name]
|
||||
params = MachineParams(**machine_params)
|
||||
machine = MachineData(name=machine_name, params=params)
|
||||
machine = MachineData(url=self.url, name=machine_name, params=params)
|
||||
return machine
|
||||
|
||||
@property
|
||||
def machine(self) -> MachineData:
|
||||
return self._machines[0]
|
||||
|
||||
def get_orig_uri(self) -> str:
|
||||
return self._orig_uri
|
||||
|
||||
def get_url(self) -> str:
|
||||
match self.url:
|
||||
case ClanUrl.LOCAL.value(path):
|
||||
return str(path)
|
||||
case ClanUrl.REMOTE.value(url):
|
||||
return url
|
||||
case _:
|
||||
raise ClanError(f"Unsupported uri components: {self.url}")
|
||||
return str(self.url)
|
||||
|
||||
@classmethod
|
||||
def from_str(
|
||||
cls, # noqa
|
||||
url: str,
|
||||
machine_name: str | None = None,
|
||||
) -> "ClanURI":
|
||||
clan_uri = ""
|
||||
if not url.startswith("clan://"):
|
||||
clan_uri += "clan://"
|
||||
|
||||
clan_uri += url
|
||||
|
||||
if machine_name:
|
||||
clan_uri += f"#{machine_name}"
|
||||
|
||||
return cls(clan_uri)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.get_orig_uri()
|
||||
|
@ -79,8 +79,8 @@ def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
||||
def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
||||
history = list_history()
|
||||
new_entries: list[HistoryEntry] = []
|
||||
for machine in list_machines(uri.get_internal()):
|
||||
new_entry = _add_maschine_to_history_list(uri.get_internal(), machine, history)
|
||||
for machine in list_machines(uri.get_url()):
|
||||
new_entry = _add_maschine_to_history_list(uri.get_url(), machine, history)
|
||||
new_entries.append(new_entry)
|
||||
write_history_file(history)
|
||||
return new_entries
|
||||
@ -89,9 +89,7 @@ def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
||||
def add_history(uri: ClanURI) -> HistoryEntry:
|
||||
user_history_file().parent.mkdir(parents=True, exist_ok=True)
|
||||
history = list_history()
|
||||
new_entry = _add_maschine_to_history_list(
|
||||
uri.get_internal(), uri.params.flake_attr, history
|
||||
)
|
||||
new_entry = _add_maschine_to_history_list(uri.get_url(), uri.machine.name, history)
|
||||
write_history_file(history)
|
||||
return new_entry
|
||||
|
||||
@ -121,9 +119,7 @@ def add_history_command(args: argparse.Namespace) -> None:
|
||||
|
||||
# takes a (sub)parser and configures it
|
||||
def register_add_parser(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument(
|
||||
"uri", type=ClanURI.from_str, help="Path to the flake", default="."
|
||||
)
|
||||
parser.add_argument("uri", type=ClanURI, help="Path to the flake", default=".")
|
||||
parser.add_argument(
|
||||
"--all", help="Add all machines", default=False, action="store_true"
|
||||
)
|
||||
|
@ -4,7 +4,7 @@ import datetime
|
||||
|
||||
from clan_cli.flakes.inspect import inspect_flake
|
||||
|
||||
from ..clan_uri import ClanURI, MachineParams
|
||||
from ..clan_uri import ClanURI
|
||||
from ..errors import ClanCmdError
|
||||
from ..locked_open import write_history_file
|
||||
from ..nix import nix_metadata
|
||||
@ -28,9 +28,9 @@ def update_history() -> list[HistoryEntry]:
|
||||
)
|
||||
uri = ClanURI.from_str(
|
||||
url=str(entry.flake.flake_url),
|
||||
params=MachineParams(machine_name=entry.flake.flake_attr),
|
||||
machine_name=entry.flake.flake_attr,
|
||||
)
|
||||
flake = inspect_flake(uri.get_url(), uri.machines[0].name)
|
||||
flake = inspect_flake(uri.get_url(), uri.machine.name)
|
||||
flake.flake_url = str(flake.flake_url)
|
||||
entry = HistoryEntry(
|
||||
flake=flake, last_used=datetime.datetime.now().isoformat()
|
||||
|
@ -55,7 +55,7 @@ def test_remote_with_clanparams() -> None:
|
||||
# Create a ClanURI object from a remote URI with parameters
|
||||
uri = ClanURI("clan://https://example.com")
|
||||
|
||||
assert uri.machines[0].name == "defaultVM"
|
||||
assert uri.machine.name == "defaultVM"
|
||||
|
||||
match uri.url:
|
||||
case ClanUrl.REMOTE.value(url):
|
||||
@ -65,11 +65,64 @@ def test_remote_with_clanparams() -> None:
|
||||
|
||||
|
||||
def test_remote_with_all_params() -> None:
|
||||
uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM")
|
||||
assert uri.machines[0].name == "myVM"
|
||||
assert uri.machines[1].name == "secondVM"
|
||||
uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM?dummy_opt=1")
|
||||
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
|
||||
|
||||
|
||||
def test_from_str_remote() -> None:
|
||||
uri = ClanURI.from_str(url="https://example.com", machine_name="myVM")
|
||||
assert uri.get_url() == "https://example.com"
|
||||
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
|
||||
|
||||
|
||||
def test_from_str_local() -> None:
|
||||
uri = ClanURI.from_str(url="~/Projects/democlan", machine_name="myVM")
|
||||
assert uri.get_url().endswith("/Projects/democlan")
|
||||
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
|
||||
|
||||
|
||||
def test_from_str_local_no_machine() -> None:
|
||||
uri = ClanURI.from_str("~/Projects/democlan")
|
||||
assert uri.get_url().endswith("/Projects/democlan")
|
||||
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
|
||||
|
||||
|
||||
def test_from_str_local_no_machine2() -> None:
|
||||
uri = ClanURI.from_str("~/Projects/democlan#syncthing-peer1")
|
||||
assert uri.get_url().endswith("/Projects/democlan")
|
||||
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
|
||||
|
@ -6,7 +6,7 @@ from cli import Cli
|
||||
from fixtures_flakes import FlakeForTest
|
||||
from pytest import CaptureFixture
|
||||
|
||||
from clan_cli.clan_uri import ClanParameters, ClanURI
|
||||
from clan_cli.clan_uri import ClanURI
|
||||
from clan_cli.dirs import user_history_file
|
||||
from clan_cli.history.add import HistoryEntry
|
||||
|
||||
@ -19,8 +19,7 @@ def test_history_add(
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
cli = Cli()
|
||||
params = ClanParameters(flake_attr="vm1")
|
||||
uri = ClanURI.from_path(test_flake_with_core.path, params=params)
|
||||
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
|
||||
cmd = [
|
||||
"history",
|
||||
"add",
|
||||
@ -40,8 +39,7 @@ def test_history_list(
|
||||
test_flake_with_core: FlakeForTest,
|
||||
) -> None:
|
||||
cli = Cli()
|
||||
params = ClanParameters(flake_attr="vm1")
|
||||
uri = ClanURI.from_path(test_flake_with_core.path, params=params)
|
||||
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
|
||||
cmd = [
|
||||
"history",
|
||||
"list",
|
||||
|
@ -13,7 +13,7 @@ from typing import IO, ClassVar
|
||||
|
||||
import gi
|
||||
from clan_cli import vms
|
||||
from clan_cli.clan_uri import ClanScheme, ClanURI
|
||||
from clan_cli.clan_uri import ClanURI, ClanUrl
|
||||
from clan_cli.history.add import HistoryEntry
|
||||
from clan_cli.machines.machines import Machine
|
||||
|
||||
@ -116,12 +116,12 @@ class VMObject(GObject.Object):
|
||||
url=self.data.flake.flake_url, flake_attr=self.data.flake.flake_attr
|
||||
)
|
||||
match uri.scheme:
|
||||
case ClanScheme.LOCAL.value(path):
|
||||
case ClanUrl.LOCAL.value(path):
|
||||
self.machine = Machine(
|
||||
name=self.data.flake.flake_attr,
|
||||
flake=path, # type: ignore
|
||||
)
|
||||
case ClanScheme.REMOTE.value(url):
|
||||
case ClanUrl.REMOTE.value(url):
|
||||
self.machine = Machine(
|
||||
name=self.data.flake.flake_attr,
|
||||
flake=url, # type: ignore
|
||||
|
@ -62,8 +62,8 @@ class JoinList:
|
||||
cls._instance = cls.__new__(cls)
|
||||
cls.list_store = Gio.ListStore.new(JoinValue)
|
||||
|
||||
# Rerendering the join list every time an item changes in the clan_store
|
||||
ClanStore.use().register_on_deep_change(cls._instance._rerender_join_list)
|
||||
|
||||
return cls._instance
|
||||
|
||||
def _rerender_join_list(
|
||||
@ -83,7 +83,9 @@ class JoinList:
|
||||
"""
|
||||
|
||||
value = JoinValue(uri)
|
||||
if value.url.get_id() in [item.url.get_id() for item in self.list_store]:
|
||||
if value.url.machine.get_id() in [
|
||||
item.url.machine.get_id() for item in self.list_store
|
||||
]:
|
||||
log.info(f"Join request already exists: {value.url}. Ignoring.")
|
||||
return
|
||||
|
||||
|
@ -57,7 +57,7 @@ class ClanStore:
|
||||
store: "GKVStore", position: int, removed: int, added: int
|
||||
) -> None:
|
||||
if added > 0:
|
||||
store.register_on_change(on_vmstore_change)
|
||||
store.values()[position].register_on_change(on_vmstore_change)
|
||||
callback(store, position, removed, added)
|
||||
|
||||
self.clan_store.register_on_change(on_clanstore_change)
|
||||
@ -111,10 +111,11 @@ class ClanStore:
|
||||
del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
|
||||
|
||||
def get_vm(self, uri: ClanURI) -> None | VMObject:
|
||||
clan = self.clan_store.get(uri.get_internal())
|
||||
if clan is None:
|
||||
vm_store = self.clan_store.get(str(uri.url))
|
||||
if vm_store is None:
|
||||
return None
|
||||
return clan.get(uri.params.flake_attr, None)
|
||||
machine = vm_store.get(uri.machine.name, None)
|
||||
return machine
|
||||
|
||||
def get_running_vms(self) -> list[VMObject]:
|
||||
return [
|
||||
|
@ -190,8 +190,8 @@ class ClanList(Gtk.Box):
|
||||
log.debug("Rendering join row for %s", join_val.url)
|
||||
|
||||
row = Adw.ActionRow()
|
||||
row.set_title(join_val.url.params.flake_attr)
|
||||
row.set_subtitle(join_val.url.get_internal())
|
||||
row.set_title(join_val.url.machine.name)
|
||||
row.set_subtitle(str(join_val.url))
|
||||
row.add_css_class("trust")
|
||||
|
||||
vm = ClanStore.use().get_vm(join_val.url)
|
||||
@ -204,7 +204,7 @@ class ClanList(Gtk.Box):
|
||||
)
|
||||
|
||||
avatar = Adw.Avatar()
|
||||
avatar.set_text(str(join_val.url.params.flake_attr))
|
||||
avatar.set_text(str(join_val.url.machine.name))
|
||||
avatar.set_show_initials(True)
|
||||
avatar.set_size(50)
|
||||
row.add_prefix(avatar)
|
||||
@ -229,7 +229,7 @@ class ClanList(Gtk.Box):
|
||||
|
||||
def on_join_request(self, source: Any, url: str) -> None:
|
||||
log.debug("Join request: %s", url)
|
||||
clan_uri = ClanURI.from_str(url)
|
||||
clan_uri = ClanURI(url)
|
||||
JoinList.use().push(clan_uri, self.on_after_join)
|
||||
|
||||
def on_after_join(self, source: JoinValue) -> None:
|
||||
|
Loading…
Reference in New Issue
Block a user