from textwrap import dedent
import click
import gql
from gql.transport.exceptions import TransportQueryError
from latch_sdk_gql.execute import execute
from latch.ldata._transfer.node import get_node_data as _get_node_data
from latch.ldata.type import LDataNodeType
from latch_cli.services.cp.glob import expand_pattern
from latch_cli.utils.path import get_name_from_path, get_path_error, is_remote_path
[docs]def move(
src: str,
dest: str,
*,
no_glob: bool = False,
):
if not is_remote_path(src) or not is_remote_path(dest):
click.secho(
f"`latch mv` cannot be used for local file operations. Please make sure"
f" all of your input paths are remote (beginning with `latch://`)",
fg="red",
)
raise click.exceptions.Exit(1)
if no_glob:
srcs = [src]
else:
srcs = expand_pattern(src)
if len(srcs) == 0:
click.echo(f"Could not find any files that match pattern {src}. Exiting.")
raise click.exceptions.Exit(0)
try:
node_data = _get_node_data(*srcs, dest, allow_resolve_to_parent=True)
except Exception as e:
click.echo(str(e))
raise click.exceptions.Exit(1) from e
dest_data = node_data.data[dest]
acc_id = node_data.acc_id
multi_src = len(srcs) > 1
if not (dest_data.exists() or dest_data.is_direct_parent()):
click.echo(get_path_error(dest, "not found", acc_id))
raise click.exceptions.Exit(1)
elif multi_src and not dest_data.exists():
click.secho(
f"Remote destination {dest} does not exist. Cannot move multiple source"
" files to a destination that does not exist.",
fg="red",
)
raise click.exceptions.Exit(1)
elif multi_src and dest_data.type in {LDataNodeType.obj, LDataNodeType.link}:
click.secho(
f"Remote destination {dest} is not a directory. Cannot move multiple source"
" files to a destination that is not a directory.",
fg="red",
)
raise click.exceptions.Exit(1)
path_by_id = {v.id: k for k, v in node_data.data.items()}
for s in srcs:
src_data = node_data.data[s]
if not src_data.exists():
click.echo(get_path_error(s, "not found", acc_id))
raise click.exceptions.Exit(1)
new_name = None
if dest_data.is_direct_parent():
new_name = get_name_from_path(dest)
elif dest_data.type in {LDataNodeType.obj, LDataNodeType.link}:
click.echo(get_path_error(dest, "object already exists at path.", acc_id))
raise click.exceptions.Exit(1)
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]
click.echo(get_path_error(path, "permission denied.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg == "Refusing to make node its own parent":
click.echo(get_path_error(dest, f"is a parent of {s}.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg == "Refusing to parent node to an object node":
click.echo(get_path_error(dest, f"object exists at path.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg == "Refusing to move a share link (or into a share link)":
if src_data.type is LDataNodeType.link:
path = s
else:
path = dest
click.echo(get_path_error(path, f"is a share link.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg.startswith("Refusing to move account root"):
click.echo(get_path_error(s, "is an account root.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg.startswith("Refusing to move removed node"):
click.echo(get_path_error(s, "not found.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg.startswith("Refusing to move already moved node"):
click.echo(
get_path_error(
s,
"copy in progress. Please wait until the node has finished"
" copying before moving.",
acc_id,
)
)
raise click.exceptions.Exit(1) from e
elif msg == "Conflicting object in destination":
click.echo(get_path_error(dest, "object exists at path.", acc_id))
raise click.exceptions.Exit(1) from e
elif msg.startswith("Refusing to do noop move"):
click.echo(get_path_error(dest, "cannot move node to itself.", acc_id))
raise click.exceptions.Exit(1) from e
else:
raise e
if len(srcs) == 1:
src_str = f'{click.style("Source: ", fg="blue")}{srcs[0]}'
else:
src_str = "\n".join(
[click.style("Sources: ", fg="blue"), *[f" {s}" for s in srcs]]
)
click.echo(dedent(f"""
{click.style("Move Succeeded.", fg="green")}
__srcs__
{click.style("Destination: ", fg="blue")}{(dest)}
""").replace("__srcs__", src_str).strip())