Delete

Delete files or entire folder structures.

Delete


Processing

The Delete brick removes files or entire folder structures from the file system, either by moving them to the recycle bin (if supported and enabled) or permanently deleting them. It processes single or multiple paths.

Inputs

source
A single file/folder path or a list of paths to delete.

Inputs Types

Input Types
source Str, Path, List

You can check the list of supported types here: Available Type Hints.

Outputs

deleted items
A list of paths (as strings) that were successfully deleted.
errors
A list of error messages for any paths that could not be deleted.

Outputs Types

Output Types
deleted items List
errors List

You can check the list of supported types here: Available Type Hints.

Options

The Delete brick contains some changeable options:

To Recycle Bin
If enabled, attempts to move files or folders to the recycle bin instead of permanently deleting them. Defaults to True.
Verbose
If enabled, logs detailed information about the deletion process, including successes and errors. Defaults to True.
import logging
import errno
import shutil
import os
import platform
from coded_flows.types import Str, Path, Tuple, Union, List
from send2trash import send2trash

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def _validate_source_input(src):
    """Validate that source input contains only strings and Path objects."""
    if isinstance(src, (str, Path)):
        return True
    elif isinstance(src, list):
        return all((isinstance(item, (str, Path)) for item in src))
    return False


def _move_to_recycle_bin(file_path, result, verbose, brick_display_name):
    """Move file/folder to recycle bin using send2trash."""
    try:
        send2trash(str(file_path))
        result["deleted_items"].append(str(file_path))
        verbose and logger.info(
            f"[{brick_display_name}] '{file_path}' was moved to recycle bin."
        )
        return True
    except Exception as e:
        verbose and logger.error(
            f"[{brick_display_name}] Failed to move '{file_path}' to recycle bin."
        )
        result["errors"].append(f"Failed to move '{file_path}' to recycle bin: {e}")
        return False


def _permanent_delete(file_path, result, verbose, brick_display_name):
    """Permanently delete file or directory."""
    try:
        if file_path.is_file() or file_path.is_symlink():
            file_path.unlink()
        elif file_path.is_dir():
            shutil.rmtree(file_path)
        else:
            raise ValueError(f"Unknown file type: {file_path}")
        result["deleted_items"].append(str(file_path))
        verbose and logger.info(
            f"[{brick_display_name}] '{file_path}' was permanently deleted."
        )
        return True
    except OSError as e:
        if e.errno == errno.EROFS:
            verbose and logger.error(
                f"[{brick_display_name}] Failed to delete '{file_path}': File system is read-only."
            )
            result["errors"].append(
                f"Failed to delete '{file_path}': File system is read-only"
            )
        else:
            verbose and logger.error(
                f"[{brick_display_name}] Failed to delete '{file_path}'."
            )
            result["errors"].append(f"Failed to delete '{file_path}': {e}")
        return False
    except PermissionError as e:
        verbose and logger.error(
            f"[{brick_display_name}] Failed to delete '{file_path}': Permission denied."
        )
        result["errors"].append(f"Failed to delete '{file_path}': Permission denied")
        return False
    except Exception as e:
        verbose and logger.error(
            f"[{brick_display_name}] Failed to delete '{file_path}': {e}"
        )
        result["errors"].append(f"Failed to delete '{file_path}': {e}")
        return False


def _delete_item(src_path, result, use_recycle_bin, verbose, brick_display_name):
    """Delete a single file or directory."""
    if use_recycle_bin:
        if _move_to_recycle_bin(src_path, result, verbose, brick_display_name):
            return
        else:
            verbose and logger.warning(
                f"[{brick_display_name}] Falling back to permanent deletion for '{src_path}'"
            )
    _permanent_delete(src_path, result, verbose, brick_display_name)


def delete_files(source: Union[Str, Path, List], options=None) -> Tuple[List, List]:
    brick_display_name = "Delete"
    options = options or {}
    verbose = options.get("verbose", True)
    use_recycle_bin = options.get("use_recycle_bin", True)
    result = {"deleted_items": [], "errors": []}
    if not _validate_source_input(source):
        verbose and logger.error(
            f"[{brick_display_name}] Invalid source input. Must be Str, Path, or list of Str/Path objects."
        )
        raise ValueError(
            "Invalid source input. Must be Str, Path, or list of Str/Path objects."
        )
    if isinstance(source, (str, Path)):
        source_paths = [Path(source)]
    else:
        source_paths = source
    for source_item in source_paths:
        src_path = Path(source_item)
        if not src_path.exists():
            result["errors"].append(f"Source path does not exist: '{source_item}'.")
            verbose and logger.error(
                f"[{brick_display_name}] Source path does not exist: '{source_item}'."
            )
            continue
        _delete_item(src_path, result, use_recycle_bin, verbose, brick_display_name)
    deleted_items = result["deleted_items"]
    errors = result["errors"]
    return (deleted_items, errors)

Brick Info

version v0.1.4
python 3.10, 3.11, 3.12, 3.13
requirements
  • send2trash