Source code for latch_cli.services.move

import click
import gql
from gql.transport.exceptions import TransportQueryError
from latch_sdk_gql.execute import execute

from latch_cli.services.cp.ldata_utils import LDataNodeType, get_node_data
from latch_cli.utils.path import get_name_from_path, get_path_error, is_remote_path


[docs]def move( src: str, dest: str, ): if not is_remote_path(src) or not is_remote_path(dest): raise ValueError( f"`latch mv` cannot be used for local file operations. Please make sure" f" both of your input paths are remote (beginning with `latch://`)" ) node_data = get_node_data(src, dest, allow_resolve_to_parent=True) src_data = node_data.data[src] dest_data = node_data.data[dest] acc_id = node_data.acc_id path_by_id = {v.id: k for k, v in node_data.data.items()} if src_data.is_parent: raise get_path_error(src, "not found", acc_id) new_name = None if dest_data.is_parent: new_name = get_name_from_path(dest) elif dest_data.type in {LDataNodeType.obj, LDataNodeType.link}: raise get_path_error(dest, "object already exists at path.", acc_id) try: execute( gql.gql(""" mutation Move( $argNode: BigInt! $argDestParent: BigInt! $argNewName: String ) { ldataMove( input: { argNode: $argNode argDestParent: $argDestParent argNewName: $argNewName } ) { clientMutationId } }"""), { "argNode": src_data.id, "argDestParent": dest_data.id, "argNewName": new_name, }, ) except TransportQueryError as e: if e.errors is None or len(e.errors) == 0: raise e msg: str = e.errors[0]["message"] if msg.startswith("Permission denied on node"): node_id = msg.rsplit(" ", 1)[1] path = path_by_id[node_id] raise get_path_error(path, "permission denied.", acc_id) from e elif msg == "Refusing to make node its own parent": raise get_path_error(dest, f"is a parent of {src}.", acc_id) from e elif msg == "Refusing to parent node to an object node": raise get_path_error(dest, f"object exists at path.", acc_id) from e elif msg == "Refusing to move a share link (or into a share link)": if src_data.type is LDataNodeType.link: path = src else: path = dest raise get_path_error(path, f"is a share link.", acc_id) from e elif msg.startswith("Refusing to move account root"): raise get_path_error(src, "is an account root.", acc_id) from e elif msg.startswith("Refusing to move removed node"): raise get_path_error(src, "not found.", acc_id) from e elif msg.startswith("Refusing to move already moved node"): raise get_path_error( src, "copy in progress. Please wait until the node has finished copying" " before moving.", acc_id, ) from e elif msg == "Conflicting object in destination": raise get_path_error(dest, "object exists at path.", acc_id) from e elif msg.startswith("Refusing to do noop move"): raise get_path_error(dest, "cannot move node to itself.", acc_id) from e else: raise e click.echo(f""" {click.style("Move Succeeded.", fg="green")} {click.style("Source: ", fg="blue")}{(src)} {click.style("Destination: ", fg="blue")}{(dest)}""")