import pickle import sys import io import os import json import uuid import logging import re # NOQA import builtins # NOQA import datetime # NOQA import decimal # NOQA import orderly_set # NOQA import collections # NOQA import fractions import ipaddress import base64 from copy import deepcopy, copy from functools import partial from collections.abc import Mapping, KeysView from typing import ( Callable, Optional, Union, overload, Literal, Any, ) from deepdiff.helper import ( strings, get_type, TEXT_VIEW, TREE_VIEW, np_float32, np_float64, np_int32, np_int64, np_ndarray, Opcode, SetOrdered, pydantic_base_model_type, PydanticBaseModel, NotPresent, ipranges, ) from deepdiff.model import DeltaResult try: import orjson except ImportError: # pragma: no cover. orjson = None logger = logging.getLogger(__name__) class UnsupportedFormatErr(TypeError): pass NONE_TYPE = type(None) CSV_HEADER_MAX_CHUNK_SIZE = 2048 # The chunk needs to be big enough that covers a couple of rows of data. MODULE_NOT_FOUND_MSG = 'DeepDiff Delta did not find {} in your modules. Please make sure it is already imported.' FORBIDDEN_MODULE_MSG = "Module '{}' is forbidden. You need to explicitly pass it by passing a safe_to_import parameter" DELTA_IGNORE_ORDER_NEEDS_REPETITION_REPORT = 'report_repetition must be set to True when ignore_order is True to create the delta object.' DELTA_ERROR_WHEN_GROUP_BY = 'Delta can not be made when group_by is used since the structure of data is modified from the original form.' SAFE_TO_IMPORT = frozenset({ 'builtins.range', 'builtins.complex', 'builtins.set', 'builtins.frozenset', 'builtins.slice', 'builtins.str', 'builtins.bytes', 'builtins.list', 'builtins.tuple', 'builtins.int', 'builtins.float', 'builtins.dict', 'builtins.bool', 'builtins.bin', 'builtins.None', 'datetime.datetime', 'datetime.time', 'datetime.timedelta', 'decimal.Decimal', 'fractions.Fraction', 'uuid.UUID', 'orderly_set.sets.OrderedSet', 'orderly_set.sets.OrderlySet', 'orderly_set.sets.StableSetEq', 'deepdiff.helper.SetOrdered', 'collections.namedtuple', 'collections.OrderedDict', 're.Pattern', 'deepdiff.helper.Opcode', 'ipaddress.IPv4Interface', 'ipaddress.IPv6Interface', 'ipaddress.IPv4Network', 'ipaddress.IPv6Network', 'ipaddress.IPv4Address', 'ipaddress.IPv6Address', 'collections.abc.KeysView', }) TYPE_STR_TO_TYPE = { 'range': range, 'complex': complex, 'set': set, 'frozenset': frozenset, 'slice': slice, 'str': str, 'bytes': bytes, 'list': list, 'tuple': tuple, 'int': int, 'float': float, 'dict': dict, 'bool': bool, 'bin': bin, 'None': None, 'NoneType': None, 'datetime': datetime.datetime, 'time': datetime.time, 'timedelta': datetime.timedelta, 'Decimal': decimal.Decimal, 'SetOrdered': SetOrdered, 'namedtuple': collections.namedtuple, 'OrderedDict': collections.OrderedDict, 'Pattern': re.Pattern, 'iprange': str, 'IPv4Address': ipaddress.IPv4Address, 'IPv6Address': ipaddress.IPv6Address, 'KeysView': list, } class ModuleNotFoundError(ImportError): """ Raised when the module is not found in sys.modules """ pass class ForbiddenModule(ImportError): """ Raised when a module is not explicitly allowed to be imported """ pass class SerializationMixin: def to_json_pickle(self): """ :ref:`to_json_pickle_label` Get the json pickle of the diff object. Unless you need all the attributes and functionality of DeepDiff, running to_json() is the safer option that json pickle. """ try: import jsonpickle copied = self.copy() # type: ignore return jsonpickle.encode(copied) except ImportError: # pragma: no cover. Json pickle is getting deprecated. logger.error('jsonpickle library needs to be installed in order to run to_json_pickle') # pragma: no cover. Json pickle is getting deprecated. @classmethod def from_json_pickle(cls, value): """ :ref:`from_json_pickle_label` Load DeepDiff object with all the bells and whistles from the json pickle dump. Note that json pickle dump comes from to_json_pickle """ try: import jsonpickle return jsonpickle.decode(value) except ImportError: # pragma: no cover. Json pickle is getting deprecated. logger.error('jsonpickle library needs to be installed in order to run from_json_pickle') # pragma: no cover. Json pickle is getting deprecated. def to_json(self, default_mapping: Optional[dict]=None, force_use_builtin_json=False, verbose_level: Optional[int]=None, **kwargs): """ Dump json of the text view. **Parameters** default_mapping : dictionary(optional), a dictionary of mapping of different types to json types. by default DeepDiff converts certain data types. For example Decimals into floats so they can be exported into json. If you have a certain object type that the json serializer can not serialize it, please pass the appropriate type conversion through this dictionary. force_use_builtin_json: Boolean, default = False When True, we use Python's builtin Json library for serialization, even if Orjson is installed. verbose_level: int, default=None Override the verbose_level for the serialized output. See to_dict() for details. kwargs: Any other kwargs you pass will be passed on to Python's json.dumps() **Example** Serialize custom objects >>> class A: ... pass ... >>> class B: ... pass ... >>> t1 = A() >>> t2 = B() >>> ddiff = DeepDiff(t1, t2) >>> ddiff.to_json() TypeError: We do not know how to convert <__main__.A object at 0x10648> of type for json serialization. Please pass the default_mapping parameter with proper mapping of the object to a basic python type. >>> default_mapping = {A: lambda x: 'obj A', B: lambda x: 'obj B'} >>> ddiff.to_json(default_mapping=default_mapping) '{"type_changes": {"root": {"old_type": "A", "new_type": "B", "old_value": "obj A", "new_value": "obj B"}}}' """ dic = self.to_dict(verbose_level=verbose_level) return json_dumps( dic, default_mapping=default_mapping, force_use_builtin_json=force_use_builtin_json, **kwargs, ) def to_dict(self, verbose_level: Optional[int]=None) -> dict: """ Convert the result to a python dictionary. **Parameters** verbose_level: int, default=None Override the verbose_level for the serialized output. When None, the behavior depends on the original view: - If the original view is 'text', the verbose_level from DeepDiff initialization is used. - If the original view is 'tree', verbose_level=2 is used to provide the most detailed output. Valid values are 0, 1, or 2. """ if verbose_level is not None and verbose_level not in {0, 1, 2}: raise ValueError('verbose_level should be 0, 1, or 2.') if verbose_level is None: if self.view == TREE_VIEW: # type: ignore verbose_level = 2 else: verbose_level = self.verbose_level # type: ignore return dict(self._get_view_results(TEXT_VIEW, verbose_level=verbose_level)) # type: ignore def _to_delta_dict( self, directed: bool = True, report_repetition_required: bool = True, always_include_values: bool = False, ) -> dict: """ Dump to a dictionary suitable for delta usage. Unlike to_dict, this is not dependent on the original view that the user chose to create the diff. **Parameters** directed : Boolean, default=True, whether to create a directional delta dictionary or a symmetrical Note that in the current implementation the symmetrical delta (non-directional) is ONLY used for verifying that the delta is being applied to the exact same values as what was used to generate the delta and has no other usages. If this option is set as True, then the dictionary will not have the "old_value" in the output. Otherwise it will have the "old_value". "old_value" is the value of the item in t1. If delta = Delta(DeepDiff(t1, t2)) then t1 + delta == t2 Note that it the items in t1 + delta might have slightly different order of items than t2 if ignore_order was set to be True in the diff object. """ if self.group_by is not None: # type: ignore raise ValueError(DELTA_ERROR_WHEN_GROUP_BY) if directed and not always_include_values: _iterable_opcodes = {} # type: ignore for path, op_codes in self._iterable_opcodes.items(): # type: ignore _iterable_opcodes[path] = [] for op_code in op_codes: new_op_code = Opcode( tag=op_code.tag, t1_from_index=op_code.t1_from_index, t1_to_index=op_code.t1_to_index, t2_from_index=op_code.t2_from_index, t2_to_index=op_code.t2_to_index, new_values=op_code.new_values, ) _iterable_opcodes[path].append(new_op_code) else: _iterable_opcodes = self._iterable_opcodes # type: ignore result = DeltaResult( tree_results=self.tree, # type: ignore ignore_order=self.ignore_order, # type: ignore always_include_values=always_include_values, _iterable_opcodes=_iterable_opcodes, ) result.remove_empty_keys() if report_repetition_required and self.ignore_order and not self.report_repetition: # type: ignore raise ValueError(DELTA_IGNORE_ORDER_NEEDS_REPETITION_REPORT) if directed: for report_key, report_value in result.items(): if isinstance(report_value, Mapping): for path, value in report_value.items(): if isinstance(value, Mapping) and 'old_value' in value: del value['old_value'] # type: ignore if self._numpy_paths: # type: ignore # Note that keys that start with '_' are considered internal to DeepDiff # and will be omitted when counting distance. (Look inside the distance module.) result['_numpy_paths'] = self._numpy_paths # type: ignore if self.iterable_compare_func: # type: ignore result['_iterable_compare_func_was_used'] = True return deepcopy(dict(result)) def pretty(self, prefix: Optional[Union[str, Callable]]=None): """ The pretty human readable string output for the diff object regardless of what view was used to generate the diff. prefix can be a callable or a string or None. Example: >>> t1={1,2,4} >>> t2={2,3} >>> print(DeepDiff(t1, t2).pretty()) Item root[3] added to set. Item root[4] removed from set. Item root[1] removed from set. """ result = [] if prefix is None: prefix = '' keys = sorted(self.tree.keys()) # type: ignore # sorting keys to guarantee constant order across python versions. for key in keys: for item_key in self.tree[key]: # type: ignore result += [pretty_print_diff(item_key)] if callable(prefix): return "\n".join(f"{prefix(diff=self)}{r}" for r in result) return "\n".join(f"{prefix}{r}" for r in result) # Maximum size allowed for integer arguments to constructors that allocate # memory proportional to the argument (e.g. bytes(n), bytearray(n)). # This prevents denial-of-service via crafted pickle payloads. (CVE-2026-33155) _MAX_ALLOC_SIZE = 128 * 1024 * 1024 # 128 MB # Callables where an integer argument directly controls memory allocation size. _SIZE_SENSITIVE_CALLABLES = frozenset({bytes, bytearray}) class _SafeConstructor: """Wraps a type constructor to prevent excessive memory allocation via the REDUCE opcode.""" __slots__ = ('_wrapped',) def __init__(self, wrapped): self._wrapped = wrapped def __call__(self, *args, **kwargs): for arg in args: if isinstance(arg, int) and arg > _MAX_ALLOC_SIZE: raise pickle.UnpicklingError( "Refusing to create {}() with size {}: " "exceeds the maximum allowed size of {} bytes. " "This could be a denial-of-service attack payload.".format( self._wrapped.__name__, arg, _MAX_ALLOC_SIZE ) ) return self._wrapped(*args, **kwargs) class _RestrictedUnpickler(pickle.Unpickler): def __init__(self, *args, **kwargs): self.safe_to_import = kwargs.pop('safe_to_import', None) if self.safe_to_import: if isinstance(self.safe_to_import, strings): self.safe_to_import = set([self.safe_to_import]) elif isinstance(self.safe_to_import, (set, frozenset)): pass else: self.safe_to_import = set(self.safe_to_import) self.safe_to_import = self.safe_to_import | SAFE_TO_IMPORT else: self.safe_to_import = SAFE_TO_IMPORT super().__init__(*args, **kwargs) def find_class(self, module, name): # Only allow safe classes from self.safe_to_import. module_dot_class = '{}.{}'.format(module, name) if module_dot_class in self.safe_to_import: try: module_obj = sys.modules[module] except KeyError: raise ModuleNotFoundError(MODULE_NOT_FOUND_MSG.format(module_dot_class)) from None cls = getattr(module_obj, name) # Wrap size-sensitive callables to prevent DoS via large allocations if cls in _SIZE_SENSITIVE_CALLABLES: return _SafeConstructor(cls) return cls # Forbid everything else. raise ForbiddenModule(FORBIDDEN_MODULE_MSG.format(module_dot_class)) from None def persistent_load(self, pid): if pid == "<>": return type(None) class _RestrictedPickler(pickle.Pickler): def persistent_id(self, obj): if obj is NONE_TYPE: # NOQA return "<>" return None def pickle_dump(obj, file_obj=None, protocol=4): """ **pickle_dump** Dumps the obj into pickled content. **Parameters** obj : Any python object file_obj : (Optional) A file object to dump the contents into **Returns** If file_obj is passed the return value will be None. It will write the object's pickle contents into the file. However if no file_obj is passed, then it will return the pickle serialization of the obj in the form of bytes. """ file_obj_passed = bool(file_obj) file_obj = file_obj or io.BytesIO() _RestrictedPickler(file_obj, protocol=protocol, fix_imports=False).dump(obj) if not file_obj_passed: return file_obj.getvalue() def pickle_load(content=None, file_obj=None, safe_to_import=None): """ **pickle_load** Load the pickled content. content should be a bytes object. **Parameters** content : Bytes of pickled object. file_obj : A file object to load the content from safe_to_import : A set of modules that needs to be explicitly allowed to be loaded. Example: {'mymodule.MyClass', 'decimal.Decimal'} Note that this set will be added to the basic set of modules that are already allowed. The set of what is already allowed can be found in deepdiff.serialization.SAFE_TO_IMPORT **Returns** A delta object that can be added to t1 to recreate t2. **Examples** Importing >>> from deepdiff import DeepDiff, Delta >>> from pprint import pprint """ if not content and not file_obj: raise ValueError('Please either pass the content or the file_obj to pickle_load.') if isinstance(content, str): content = content.encode('utf-8') if content: file_obj = io.BytesIO(content) return _RestrictedUnpickler(file_obj, safe_to_import=safe_to_import).load() def _get_pretty_form_text(verbose_level): pretty_form_texts = { "type_changes": "Type of {diff_path} changed from {type_t1} to {type_t2} and value changed from {val_t1} to {val_t2}.", "values_changed": "Value of {diff_path} changed from {val_t1} to {val_t2}.", "dictionary_item_added": "Item {diff_path} added to dictionary.", "dictionary_item_removed": "Item {diff_path} removed from dictionary.", "iterable_item_added": "Item {diff_path} added to iterable.", "iterable_item_removed": "Item {diff_path} removed from iterable.", "attribute_added": "Attribute {diff_path} added.", "attribute_removed": "Attribute {diff_path} removed.", "set_item_added": "Item root[{val_t2}] added to set.", "set_item_removed": "Item root[{val_t1}] removed from set.", "repetition_change": "Repetition change for item {diff_path}.", } if verbose_level == 2: pretty_form_texts.update( { "dictionary_item_added": "Item {diff_path} ({val_t2}) added to dictionary.", "dictionary_item_removed": "Item {diff_path} ({val_t1}) removed from dictionary.", "iterable_item_added": "Item {diff_path} ({val_t2}) added to iterable.", "iterable_item_removed": "Item {diff_path} ({val_t1}) removed from iterable.", "attribute_added": "Attribute {diff_path} ({val_t2}) added.", "attribute_removed": "Attribute {diff_path} ({val_t1}) removed.", } ) return pretty_form_texts def pretty_print_diff(diff): type_t1 = get_type(diff.t1).__name__ type_t2 = get_type(diff.t2).__name__ val_t1 = '"{}"'.format(str(diff.t1)) if type_t1 == "str" else str(diff.t1) val_t2 = '"{}"'.format(str(diff.t2)) if type_t2 == "str" else str(diff.t2) diff_path = diff.path(root='root') return _get_pretty_form_text(diff.verbose_level).get(diff.report_type, "").format( diff_path=diff_path, type_t1=type_t1, type_t2=type_t2, val_t1=val_t1, val_t2=val_t2) def load_path_content(path, file_type=None): """ Loads and deserializes the content of the path. """ if file_type is None: file_type = path.split('.')[-1] if file_type == 'json': with open(path, 'r') as the_file: content = json_loads(the_file.read()) elif file_type in {'yaml', 'yml'}: try: import yaml except ImportError: # pragma: no cover. raise ImportError('Pyyaml needs to be installed.') from None # pragma: no cover. with open(path, 'r') as the_file: content = yaml.safe_load(the_file) elif file_type == 'toml': try: if sys.version_info >= (3, 11): import tomllib as tomli else: import tomli except ImportError: # pragma: no cover. raise ImportError('On python<=3.10 tomli needs to be installed.') from None # pragma: no cover. with open(path, 'rb') as the_file: content = tomli.load(the_file) elif file_type == 'pickle': with open(path, 'rb') as the_file: content = the_file.read() content = pickle_load(content) elif file_type in {'csv', 'tsv'}: try: import clevercsv # type: ignore content = clevercsv.read_dicts(path) except ImportError: # pragma: no cover. import csv with open(path, 'r') as the_file: content = list(csv.DictReader(the_file)) logger.info(f"NOTE: CSV content was empty in {path}") # Everything in csv is string but we try to automatically convert any numbers we find for row in content: for key, value in row.items(): value = value.strip() for type_ in [int, float, complex]: try: value = type_(value) except Exception: pass else: row[key] = value break else: raise UnsupportedFormatErr(f'Only json, yaml, toml, csv, tsv and pickle are supported.\n' f' The {file_type} extension is not known.') return content def save_content_to_path(content, path, file_type=None, keep_backup=True): """ Saves and serializes the content of the path. """ backup_path = f"{path}.bak" os.rename(path, backup_path) try: _save_content( content=content, path=path, file_type=file_type, keep_backup=keep_backup) except Exception: os.rename(backup_path, path) raise else: if not keep_backup: os.remove(backup_path) def _save_content(content, path, file_type, keep_backup=True): if file_type == 'json': with open(path, 'w') as the_file: content = json_dumps(content) the_file.write(content) # type: ignore elif file_type in {'yaml', 'yml'}: try: import yaml except ImportError: # pragma: no cover. raise ImportError('Pyyaml needs to be installed.') from None # pragma: no cover. with open(path, 'w') as the_file: content = yaml.safe_dump(content, stream=the_file) elif file_type == 'toml': try: import tomli_w except ImportError: # pragma: no cover. raise ImportError('Tomli-w needs to be installed.') from None # pragma: no cover. with open(path, 'wb') as the_file: content = tomli_w.dump(content, the_file) elif file_type == 'pickle': with open(path, 'wb') as the_file: content = pickle_dump(content, file_obj=the_file) elif file_type in {'csv', 'tsv'}: try: import clevercsv # type: ignore dict_writer = clevercsv.DictWriter except ImportError: # pragma: no cover. import csv dict_writer = csv.DictWriter with open(path, 'w', newline='') as csvfile: fieldnames = list(content[0].keys()) writer = dict_writer(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerows(content) else: raise UnsupportedFormatErr('Only json, yaml, toml, csv, tsv and pickle are supported.\n' f' The {file_type} extension is not known.') return content def _serialize_decimal(value): if value.as_tuple().exponent == 0: return int(value) else: return float(value) def _serialize_fraction(value): if value.denominator == 1: return value.numerator else: return float(value) def _serialize_tuple(value): if hasattr(value, '_asdict'): # namedtuple return value._asdict() return value def _serialize_bytes(value): """ Serialize bytes to JSON-compatible format. First tries UTF-8 decoding for backward compatibility. Falls back to base64 encoding for binary data. """ try: return value.decode('utf-8') except UnicodeDecodeError: return base64.b64encode(value).decode('ascii') JSON_CONVERTOR = { decimal.Decimal: _serialize_decimal, fractions.Fraction: _serialize_fraction, SetOrdered: list, orderly_set.StableSetEq: list, set: list, type: lambda x: x.__name__, bytes: _serialize_bytes, datetime.datetime: lambda x: x.isoformat(), uuid.UUID: lambda x: str(x), np_float32: float, np_float64: float, np_int32: int, np_int64: int, np_ndarray: lambda x: x.tolist(), tuple: _serialize_tuple, Mapping: dict, NotPresent: str, ipranges: str, memoryview: lambda x: x.tobytes(), KeysView: list, } if PydanticBaseModel is not pydantic_base_model_type: JSON_CONVERTOR[PydanticBaseModel] = lambda x: x.model_dump() def json_convertor_default(default_mapping=None): if default_mapping: _convertor_mapping = JSON_CONVERTOR.copy() _convertor_mapping.update(default_mapping) else: _convertor_mapping = JSON_CONVERTOR def _convertor(obj): for original_type, convert_to in _convertor_mapping.items(): if isinstance(obj, original_type): return convert_to(obj) # This is to handle reverse() which creates a generator of type list_reverseiterator if obj.__class__.__name__ == 'list_reverseiterator': return list(copy(obj)) # 3) gather @property values by scanning __class__.__dict__ and bases props = {} for cls in obj.__class__.__mro__: for name, descriptor in cls.__dict__.items(): if isinstance(descriptor, property) and not name.startswith('_'): try: props[name] = getattr(obj, name) except Exception: # skip properties that error out pass if props: return props # 4) fallback: public __dict__ entries if hasattr(obj, '__dict__'): return { k: v for k, v in vars(obj).items() if not k.startswith('_') } # 5) give up raise TypeError( f"Don't know how to JSON-serialize {obj!r} " f"(type {type(obj).__name__}); " "consider adding it to default_mapping." ) return _convertor class JSONDecoder(json.JSONDecoder): def __init__(self, *args, **kwargs): json.JSONDecoder.__init__(self, object_hook=self.object_hook, *args, **kwargs) def object_hook(self, obj): # type: ignore if 'old_type' in obj and 'new_type' in obj: for type_key in ('old_type', 'new_type'): type_str = obj[type_key] obj[type_key] = TYPE_STR_TO_TYPE.get(type_str, type_str) return obj @overload def json_dumps( item: Any, **kwargs, ) -> str: ... @overload def json_dumps( item: Any, default_mapping:Optional[dict], force_use_builtin_json: bool, return_bytes:Literal[True], **kwargs, ) -> bytes: ... @overload def json_dumps( item: Any, default_mapping:Optional[dict], force_use_builtin_json: bool, return_bytes:Literal[False], **kwargs, ) -> str: ... _INT64_MAX = 9223372036854775807 _INT64_MIN = -9223372036854775808 def _convert_oversized_ints(obj): """Recursively convert integers exceeding 64-bit range to strings. orjson cannot serialize integers outside the signed 64-bit range.""" if isinstance(obj, bool): return obj if isinstance(obj, int) and (obj > _INT64_MAX or obj < _INT64_MIN): return str(obj) if isinstance(obj, dict): return {k: _convert_oversized_ints(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): converted = [_convert_oversized_ints(v) for v in obj] if hasattr(obj, '_fields'): # NamedTuple: reconstruct using keyword arguments return type(obj)(**dict(zip(obj._fields, converted))) # type: ignore[attr-defined] return type(obj)(converted) return obj def json_dumps( item: Any, default_mapping:Optional[dict]=None, force_use_builtin_json: bool = False, return_bytes: bool = False, **kwargs, ) -> Union[str, bytes]: """ Dump json with extra details that are not normally json serializable parameters ---------- force_use_builtin_json: Boolean, default = False When True, we use Python's builtin Json library for serialization, even if Orjson is installed. """ if orjson and not force_use_builtin_json: indent = kwargs.pop('indent', None) kwargs['option'] = orjson.OPT_NON_STR_KEYS | orjson.OPT_SERIALIZE_NUMPY if indent: kwargs['option'] |= orjson.OPT_INDENT_2 if 'sort_keys' in kwargs: raise TypeError( "orjson does not accept the sort_keys parameter. " "If you need to pass sort_keys, set force_use_builtin_json=True " "to use Python's built-in json library instead of orjson.") try: result = orjson.dumps( item, default=json_convertor_default(default_mapping=default_mapping), **kwargs) except TypeError as e: if 'Integer exceeds 64-bit range' in str(e): item = _convert_oversized_ints(item) result = orjson.dumps( item, default=json_convertor_default(default_mapping=default_mapping), **kwargs) else: raise if return_bytes: return result return result.decode(encoding='utf-8') else: result = json.dumps( item, default=json_convertor_default(default_mapping=default_mapping), **kwargs) if return_bytes: return result.encode(encoding='utf-8') return result json_loads = partial(json.loads, cls=JSONDecoder)