From 5c3be8803afa31b684b1e804485bdc8e0f1209f6 Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Tue, 20 Oct 2026 13:56:40 -0400 Subject: [PATCH] Improved file format error handling --- .../write_or_read_from_json.py | 139 +++++++++--------- 1 file changed, 67 insertions(+), 72 deletions(-) diff --git a/core/utils/basic_operations/write_or_read_from_json.py b/core/utils/basic_operations/write_or_read_from_json.py index d671f7e..b771435 100644 --- a/core/utils/basic_operations/write_or_read_from_json.py +++ b/core/utils/basic_operations/write_or_read_from_json.py @@ -4,77 +4,67 @@ import json import os from typing import Any import tempfile -import shutil +# import shutil from enum import Enum +from typing import Optional def write_json_to_file(data: dict, filepath: str) -> bool: + temp_path = None try: - # Create directory if it doesn't exist directory = os.path.dirname(filepath) if directory and not os.path.exists(directory): os.makedirs(directory) - # Write to a temporary file first - with tempfile.NamedTemporaryFile(mode='w', dir=directory, delete=False, suffix='.json') as f: - json.dump(data, f, indent=4) - temp_path = f.name - - # Only replace the original if successful - shutil.move(temp_path, filepath) + fd, temp_path = tempfile.mkstemp(dir=directory, suffix='.tmp') + + with os.fdopen(fd, 'w', encoding='utf-8') as f: + # Use 'default' to handle Enums and other non-serializable objects recursively + json.dump(data, f, default=lambda o: o.value if isinstance(o, Enum) else str(o)) + f.flush() + os.fsync(f.fileno()) + + os.replace(temp_path, filepath) return True - except TypeError as e: - logger.error(f"Data is not JSON serializable: {e}") - - # Attempt to sanitize Enums by extracting their .value - sanitized_data = {k: (v.value if isinstance(v, Enum) else v) for k, v in data.items()} - - # Only retry if we actually modified the data to prevent infinite recursion - if sanitized_data != data: - return write_json_to_file(sanitized_data, filepath) - + except (TypeError, ValueError) as e: + logger.error(f"JSON serialization error: {e}") return False - except FileNotFoundError: - logger.error(f"File not found: {filepath}") - return False - except IsADirectoryError: - logger.error(f"Path is a directory, not a file: {filepath}") + except (FileNotFoundError, IsADirectoryError, PermissionError) as e: + logger.error(f"File system error for {filepath}: {e}") return False except IOError as e: - logger.error(f"Error reading file {filepath}: {e}") + logger.error(f"IO Error writing file {filepath}: {e}") return False + finally: + # Only attempt to remove if temp_path was created and still exists + if temp_path and os.path.exists(temp_path): + try: + os.remove(temp_path) + except OSError: + pass -def read_entire_json(filepath: str) -> dict: +def read_entire_json(filepath: str) -> Optional[dict]: + """ + Reads a JSON file and returns the data. + Returns None if the file cannot be read or is missing. + """ try: - # Check if file exists - if not os.path.exists(filepath): - raise FileNotFoundError(f"File not found: {filepath}") + # Explicit encoding prevents UnicodeDecodeError. + with open(filepath, "r", encoding="utf-8") as f: + return json.load(f) - # Check if path is a directory - if os.path.isdir(filepath): - raise IsADirectoryError(f"Path is a directory, not a file: {filepath}") - - # Open and parse JSON file - with open(filepath, "r") as f: - data = json.load(f) - - return data - - except json.JSONDecodeError as e: - raise json.JSONDecodeError( - f"Invalid JSON in file {filepath}: {e.msg}", e.doc, e.pos - ) except FileNotFoundError: logger.error(f"File not found: {filepath}") - return False except IsADirectoryError: logger.error(f"Path is a directory, not a file: {filepath}") - return False + except json.JSONDecodeError as e: + logger.error(f"Invalid JSON in file {filepath}: {e.msg}") except IOError as e: - logger.error(f"Error reading file {filepath}: {e}") - return False + logger.error(f"IO Error reading file {filepath}: {e}") + + return None def _search_recursive(obj: Any, search_key: str) -> Any | None: @@ -98,48 +88,53 @@ def _search_recursive(obj: Any, search_key: str) -> Any | None: return None -def get_value_from_json_file(filepath: str, key: str, category: str | None = None) -> Any: + +def get_value_from_json_file(filepath: str, key: str, category: str | None = None) -> Any | None: """ Load JSON file and search for a key recursively. - - Args: - filepath: Path to the JSON file - key: The key to search for - category: (Optional) Top-level key to limit search scope. If None, searches entire document. - - Returns: - The value associated with the key - - Raises: - ValueError: If value is None, empty string, or empty collection - KeyError: If category or key is not found + Returns the value if found and valid; otherwise returns None. """ data = read_entire_json(filepath) + # System/Read error + if data is None: + # Note: read_entire_json already logged the specific reason (FileNotFound, etc.) + return None + + # Empty file + if not data: + logger.error(f"JSON file is empty: {filepath}") + return None + + # Handle category scope if category: if not isinstance(category, str): - error_msg = f"Error in get_value_from_json_file, category must be a string, not {type(category)}: {category}" - logger.error(error_msg) - raise ValueError(error_msg) + logger.error(f"Category must be a string, not {type(category)}: {category}") + return None if category not in data: - raise KeyError(f"Category '{category}' not found in JSON file") + logger.error(f"Category '{category}' not found in {filepath}") + return None subcategory = data[category] else: subcategory = data + # Search for the key value = _search_recursive(subcategory, key) + # Guard: Key not found if value is None: - raise KeyError(f"Key '{key}' not found" + (f" in category '{category}'" if category else "")) + logger.error(f"Key '{key}' not found" + (f" in category '{category}'" if category else "")) + return None - # Validation - if isinstance(value, str): - if value.strip() == "": - raise ValueError(f"Value for key '{key}' is empty or whitespace-only") - elif isinstance(value, (list, dict)): - if len(value) == 0: - raise ValueError(f"Value for key '{key}' is an empty collection") + # Guard: Validation (Empty strings or collections) + if isinstance(value, str) and value.strip() == "": + logger.error(f"Value for key '{key}' is empty or whitespace-only") + return None + + if isinstance(value, (list, dict)) and len(value) == 0: + logger.error(f"Value for key '{key}' is an empty collection") + return None return value