Source code for latch.types.utils

import re
from pathlib import Path
from typing import Optional, Union
from urllib.parse import urlparse

import gql
from latch_sdk_gql.execute import execute

[docs]def is_valid_url(raw_url: Union[str, Path]) -> bool: """A valid URL (as a source or destination of a LatchFile) must: * contain a latch or s3 scheme * contain an absolute path """ try: parsed = urlparse(str(raw_url)) except ValueError: return False if parsed.scheme not in ("latch", "s3"): return False if parsed.path != "" and not parsed.path.startswith("/"): return False return True
is_absolute_node_path = re.compile(r"^(latch)?://(?P<node_id>\d+).node(/)?$") old_style_path = re.compile(r"^(?:(?P<account_root>account_root)|(?P<mount>mount))")
[docs]def format_path(path: str) -> str: match = is_absolute_node_path.match(path) if match is None: return path node_id ="node_id") data = execute( gql.gql(""" query ldataGetPathQ($id: BigInt!) { ldataGetPath(argNodeId: $id) ldataOwner(argNodeId: $id) } """), {"id": node_id}, ) raw: Optional[str] = data["ldataGetPath"] if raw is None: return path parts = raw.split("/") match = old_style_path.match(raw) if match is None: return path if match["mount"] is not None: bucket = parts[1] key = "/".join(parts[2:]) return f"latch://{bucket}.mount/{key}" owner: Optional[str] = data["ldataOwner"] if owner is None: return path if match["account_root"] is not None: key = "/".join(parts[2:]) return f"latch://{owner}.account/{key}" return path