clan-core/pkgs/clan-cli/clan_cli/clan_uri.py

# Import the urllib.parse, enum and dataclasses modules
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:
        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

    @member
    @dataclass
    class LOCAL:
        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


# Parameters defined here will be DELETED from the nested uri
# so make sure there are no conflicts with other webservices
@dataclass
class MachineParams:
    dummy_opt: str = "dummy"


@dataclass
class MachineData:
    name: str = "defaultVM"
    params: MachineParams = dataclasses.field(default_factory=MachineParams)


# Define the ClanURI class
class ClanURI:
    _orig_uri: str
    _nested_uri: str
    _components: urllib.parse.ParseResult
    url: ClanUrl
    machines: list[MachineData]

    # Initialize the class with a clan:// URI
    def __init__(self, uri: str) -> None:
        self.machines = []

        # users might copy whitespace along with the uri
        uri = uri.strip()
        self._orig_uri = uri

        # Check if the URI starts with clan://
        # If it does, remove the clan:// prefix
        if uri.startswith("clan://"):
            self._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)

        # Replace the query string in the components with the new query string
        clean_comps = self._components._replace(
            query=self._components.query, fragment=""
        )

        # Parse the URL into a ClanUrl object
        self.url = self._parse_url(clean_comps)

        # Parse the fragment into a list of machine queries
        # Then parse every machine query into a MachineParameters object
        machine_frags = list(
            filter(lambda x: len(x) > 0, self._components.fragment.split("#"))
        )
        for machine_frag in machine_frags:
            machine = self._parse_machine_query(machine_frag)
            self.machines.append(machine)

        # If there are no machine fragments, add a default machine
        if len(machine_frags) == 0:
            self.machines.append(MachineData())

    def _parse_url(self, comps: urllib.parse.ParseResult) -> ClanUrl:
        comb = (
            comps.scheme,
            comps.netloc,
            comps.path,
            comps.params,
            comps.query,
            comps.fragment,
        )
        match comb:
            case ("file", "", path, "", "", _) | ("", "", path, "", "", _):  # type: ignore
                url = ClanUrl.LOCAL.value(Path(path).expanduser().resolve())  # type: ignore
            case _:
                url = ClanUrl.REMOTE.value(comps.geturl())  # type: ignore

        return url

    def _parse_machine_query(self, machine_frag: str) -> MachineData:
        comp = urllib.parse.urlparse(machine_frag)
        query = urllib.parse.parse_qs(comp.query)
        machine_name = comp.path

        machine_params: dict[str, Any] = {}
        for dfield in dataclasses.fields(MachineParams):
            if dfield.name in query:
                values = query[dfield.name]
                if len(values) > 1:
                    raise ClanError(f"Multiple values for parameter: {dfield.name}")
                machine_params[dfield.name] = values[0]

                # Remove the field from the query dictionary
                # clan uri and nested uri share one namespace for query parameters
                # we need to make sure there are no conflicts
                del query[dfield.name]
        params = MachineParams(**machine_params)
        machine = MachineData(name=machine_name, params=params)
        return machine

    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}")

    def __str__(self) -> str:
        return self.get_orig_uri()

    def __repr__(self) -> str:
        return f"ClanURI({self})"