dont use own node_data

Signed-off-by: Ayush Kamat <ayush@latch.bio>

latch_cli/services/cp/download/main.py

@@ -14,7 +14,6 @@
 from ....utils import get_auth_header, human_readable_time, urljoins, with_si_suffix
 from ....utils.path import normalize_path
 from ..glob import expand_pattern
-from ..utils import LDataNodeType, get_node_data
 from .worker import Work, worker
 
 http_session = requests.Session()
@@ -71,6 +70,7 @@ def download(
         raise click.exceptions.Exit(1)
 
     from latch.ldata.path import _get_node_data
+    from latch.ldata.type import LDataNodeType
 
     all_node_data = _get_node_data(*srcs)
     work_queue = asyncio.Queue[Work]()

latch_cli/services/cp/upload/main.py

@@ -11,8 +11,7 @@
 import tqdm
 
 from ....utils import human_readable_time, urljoins, with_si_suffix
-from ....utils.path import get_path_error, normalize_path
-from ..utils import LDataNodeType, get_node_data
+from ....utils.path import normalize_path
 from .worker import Work, worker
 
 
@@ -33,7 +32,10 @@ def upload(
 
     start = time.monotonic()
 
-    dest_data = get_node_data(dest)
+    from latch.ldata.path import _get_node_data
+    from latch.ldata.type import LDataNodeType
+
+    dest_data = _get_node_data(dest).data[dest]
     dest_is_dir = dest_data.type in {
         LDataNodeType.account_root,
         LDataNodeType.mount,

latch_cli/services/cp/utils.py

@@ -165,65 +165,3 @@ def _get_known_domains_for_account() -> List[str]:
     res.extend(f"{x}.mount" for x in buckets)
 
     return res
-
-
-class FinalLinkTarget(TypedDict):
-    id: str
-    name: str
-    type: str
-
-
-class LdataNode(TypedDict):
-    finalLinkTarget: FinalLinkTarget
-
-
-class LdataResolvePathToNode(TypedDict):
-    path: Optional[str]
-    ldataNode: LdataNode
-
-
-class LDataNodeType(str, Enum):
-    account_root = "account_root"
-    dir = "dir"
-    obj = "obj"
-    mount = "mount"
-    link = "link"
-    mount_gcp = "mount_gcp"
-    mount_azure = "mount_azure"
-
-
-@dataclass
-class NodeData:
-    type: LDataNodeType
-    exists: bool
-    name: str
-
-
-def get_node_data(remote: str) -> NodeData:
-    res: Optional[LdataResolvePathToNode] = execute(
-        gql.gql("""
-        query GetNodeData($argPath: String!) {
-            ldataResolvePathToNode(path: $argPath) {
-                path
-                ldataNode {
-                    finalLinkTarget {
-                        id
-                        name
-                        type
-                    }
-                }
-            }
-        }
-    """),
-        {"argPath": remote.rstrip("/")},
-    )["ldataResolvePathToNode"]
-
-    if res is None:
-        raise ValueError("unauthorized")
-
-    exists = res["path"] is None or res["path"] == ""
-
-    flt = res["ldataNode"]["finalLinkTarget"]
-    type = LDataNodeType(flt["type"].lower())
-
-    return NodeData(type, exists, flt["name"])