from apollo_utils.core.utils.singleton import SingletonMeta from typing import Any from server.constants.common import ABSENT class GlomAbsent(metaclass=SingletonMeta): pass GLOM_ABSENT = GlomAbsent() class GlomError(Exception): message: str def __init__(self, exc, path, dest_name): self.exc = exc self.path = path self.dest_name = dest_name def get_message(self): return self.message % (self.dest_name, self.path, self.exc) def __repr__(self): cn = self.__class__.__name__ return "%s(%r, %r, %r)" % (cn, self.exc, self.path, self.dest_name) class PathAccessError(GlomError): message = "Could not access %r at path %r: %r" class PathAssignError(GlomError): message = "Could not assign to %r at path %r: %r" class PathDeleteError(GlomError): message = "Could not delete from %r at path %r: %r" def glom(target: Any, spec: str, default: Any = GLOM_ABSENT) -> Any: """If you do not want to rise an error, specify another default value""" original_error = None obj = target for key_part in spec.split("."): if isinstance(obj, dict): obj = obj.get(key_part) elif isinstance(obj, (tuple, list)): try: obj = obj[int(key_part)] except (IndexError, ValueError) as ex: # ValueError is for non-numeric keys obj, original_error = default, ex else: obj = default if obj is GLOM_ABSENT and default is GLOM_ABSENT: # means we want to raise an error raise PathAccessError(original_error or KeyError(f"Path {key_part} not found in {obj}"), spec, target) return obj def delete(obj: Any, path: str, ignore_missing: bool = False) -> Any: key_parts = path.split(".") path_prefix, last_key_part = ".".join(key_parts[:-1]), key_parts[-1] original_error = None target = glom(obj, path_prefix, default=ABSENT) # do not want to raise an error if isinstance(target, (tuple, list)): try: target.pop(int(last_key_part)) except (IndexError, ValueError) as exc: # ValueError is for non-numeric keys original_error = exc elif not isinstance(target, dict) or target.pop(last_key_part, GLOM_ABSENT) is GLOM_ABSENT: original_error = KeyError(f"Path {path} not found in {obj}") if original_error and not ignore_missing: raise PathDeleteError(original_error, path, obj) return obj def assign(obj: Any, path: str, value: Any, missing=None) -> Any: key_parts = path.split(".") last_key_part = key_parts[-1] target = obj for key_part in key_parts[:-1]: if isinstance(target, dict): target[key_part] = target.get(key_part, missing() if callable(missing) else missing) target = glom(target, key_part) if isinstance(target, list): try: last_key_part = int(last_key_part) except ValueError as ex: raise PathAssignError(ex, path, obj) try: target[last_key_part] = value except (IndexError, TypeError) as ex: raise PathAssignError(ex, path, obj) return obj