works

local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py

@@ -9,7 +9,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from aws_resource_management.managers.base import ResourceManager
-from aws_resource_management.utils.aws_core_utils import create_client
+from aws_resource_management.utils.aws_core_utils import create_client, create_cache_key, safe_get_dict_value
 from aws_resource_management.utils.logging_utils import logger
 from aws_resource_management.utils.config_utils import get_config
 from aws_resource_management.utils.datetime_utils import parse_iso_datetime

local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/aws_core_utils.py

@@ -6,6 +6,8 @@
 """
 
 import configparser
+import hashlib
+import json
 import os
 import threading
 import time
@@ -606,3 +608,34 @@ def get_account_alias(account_id: str, region: Optional[str] = None, credentials
 
     # Default to account ID
     return account_id
+
+def create_cache_key(*args) -> str:
+    """Create a safe cache key from arguments, handling unhashable types."""
+    key_parts = []
+    for arg in args:
+        if isinstance(arg, dict):
+            key_parts.append(json.dumps(arg, sort_keys=True, default=str))
+        elif isinstance(arg, (list, tuple)):
+            # Handle lists that may contain unhashable types like dicts
+            try:
+                # Try to sort if all elements are comparable
+                if isinstance(arg, list) and all(not isinstance(item, (dict, list, set)) for item in arg):
+                    key_parts.append(str(sorted(arg)))
+                else:
+                    # For lists with complex types, convert to JSON
+                    key_parts.append(json.dumps(arg, sort_keys=True, default=str))
+            except TypeError:
+                # If sorting fails, convert to JSON
+                key_parts.append(json.dumps(arg, sort_keys=True, default=str))
+        else:
+            key_parts.append(str(arg) if arg is not None else "none")
+
+    # Create hash from joined parts for consistent key length
+    key_string = "|".join(key_parts)
+    return hashlib.md5(key_string.encode()).hexdigest()
+
+def safe_get_dict_value(dictionary: Dict[str, Any], key: str, default: Any = None) -> Any:
+    """Safely get a value from a dictionary, handling None dictionaries."""
+    if not isinstance(dictionary, dict):
+        return default
+    return dictionary.get(key, default)