def rmtree()

in hf3fs_utils/cli.py [0:0]

• 57 lines of code
• 21 McCabe index (conditional complexity)

def rmtree(dir_paths: List[str], expire: Optional[str], yes: bool):
    """
    Move a directory tree to the trash and set an expiration time, it will be automatically deleted after expiration

    \b
    Example:
    hf3fs_cli rmtree <path/to/remove> --expire <expire_time>

    \b
    - Use --expire [1h|3h|8h|1d|3d|7d] to specify the expiration time, the directory will be deleted after expiration.
    - Before expiration, you can restore the directory from the trash using `hf3fs_cli mv <trash_path> <target_path>`.
    - If you need to free up space immediately, you can use `hf3fs_cli rmtree <trash_path>` to delete the data in the trash immediately, this operation cannot be undone!
    - Use `ls /hf3fs/{cluster}/trash` to view the trash.
    """

    if not dir_paths:
        abort(f"Please provide the directory path to delete")

    first_path = abs_path(dir_paths[0])
    fs = get_filesystem(first_path)
    fs_trash = Trash(fs)

    clean_trash = is_relative_to(first_path, fs_trash.trash_path)
    if not clean_trash:
        if not expire:
            abort(f"Use --expire [1h|3h|8h|1d|3d|7d] to specify the expiration time")
    elif expire:
        abort(f"{first_path} is already in the trash")
    trash_cfg = TRASH_CONFIGS[expire] if not clean_trash else None

    dir_paths = [abs_path(p) for p in dir_paths]
    for dir_path in dir_paths:
        if is_relative_to(dir_path, fs_trash.trash_path) != clean_trash:
            if clean_trash:
                abort(f"{dir_path} is not in the trash")
            else:
                abort(f"{dir_path} is already in the trash")

    if clean_trash:
        if len(dir_paths) != 1:
            msg = (
                f"Immediately delete the following paths:\n"
                + "\n".join([f"- {p}" for p in dir_paths])
                + "\nThis operation cannot be undone"
            )
        else:
            msg = f"Immediately delete {dir_path}, this operation cannot be undone"
    else:
        if len(dir_paths) != 1:
            msg = (
                f"Move the following paths to the trash:\n"
                + "\n".join([f"- {p}" for p in dir_paths])
                + f"\nThey will be automatically deleted after {expire}"
            )
        else:
            msg = f"Move {dir_path} to the trash, it will be automatically deleted after {expire}"
    if not yes:
        assert click.confirm(msg, abort=True)

    for dir_path in dir_paths:
        try:
            if clean_trash:
                fs.remove(dir_path, recursive=True)
                click.echo(f"- Deleted {dir_path}")
            else:
                trash_path = fs_trash.move_to_trash(dir_path, trash_cfg)
                click.echo(f"- Trash path: {trash_path}")
        except AssertionError:
            raise
        except Exception as ex:
            abort(f"Failed to delete {dir_path}: {ex}")

    if not clean_trash:
        click.echo(
            "- Before expiration, you can use 'hf3fs_cli mv <trash_path> <target_path>' to restore, "
            "or use 'hf3fs_cli rmtree <trash_path>' to delete immediately and free up space"
        )