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