Improved file format error handling

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
+        fd, temp_path = tempfile.mkstemp(dir=directory, suffix='.tmp')
         
-        # Only replace the original if successful
-        shutil.move(temp_path, filepath)
+        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