DeepHash._hash ran normalize_value_for_hashing of every registered custom operator on every value, without ever consulting the operator's match method. An operator scoped to one type therefore rewrote values of every other type. When such an operator returns None for the values it was never meant to see, all of those values hash to the same thing and unequal iterables compare as equal under ignore_order=True: [1, 2] and [2, 3] were reported as identical. Ask the operator whether it applies before normalizing. DeepHash walks a single object at a time and has no t1/t2 pair, so the value is passed to match as both sides via a small level-like view, which keeps operators scoped by type or by path behaving as they do while diffing. Operators that define no match method are still applied to everything. Fixes #606
797 lines
33 KiB
Python
797 lines
33 KiB
Python
#!/usr/bin/env python
|
|
import logging
|
|
import datetime
|
|
import uuid
|
|
from typing import Union, Optional, Any, List, TYPE_CHECKING, Dict, Tuple, Set, Callable, Generator
|
|
from collections.abc import Iterable, MutableMapping
|
|
from collections import defaultdict
|
|
from hashlib import sha1, sha256
|
|
from pathlib import Path
|
|
from enum import Enum
|
|
import re
|
|
from deepdiff.helper import (strings, numbers, only_numbers, times, unprocessed, not_hashed, add_to_frozen_set,
|
|
convert_item_or_items_into_set_else_none, get_doc, ipranges,
|
|
convert_item_or_items_into_compiled_regexes_else_none,
|
|
get_id, type_is_subclass_of_type_group, type_in_type_group,
|
|
number_to_string, datetime_normalize, KEY_TO_VAL_STR,
|
|
get_truncate_datetime, dict_, add_root_to_paths, PydanticBaseModel,
|
|
separate_wildcard_and_exact_paths,
|
|
SetOrdered)
|
|
|
|
from deepdiff.base import Base
|
|
|
|
if TYPE_CHECKING:
|
|
from pytz.tzinfo import BaseTzInfo
|
|
import numpy as np
|
|
|
|
# Type aliases for better readability
|
|
HashableType = Union[str, int, float, bytes, bool, tuple, frozenset, type(None)]
|
|
HashResult = Union[str, Any] # Can be string hash or unprocessed marker
|
|
HashTuple = Tuple[HashResult, int] # (hash_result, count)
|
|
HashesDict = Dict[Any, Union[HashTuple, List[Any]]] # Special case for UNPROCESSED_KEY
|
|
PathType = Union[str, List[str], Set[str]]
|
|
RegexType = Union[str, re.Pattern[str], List[Union[str, re.Pattern[str]]]]
|
|
NumberToStringFunc = Callable[..., str] # More flexible for different number_to_string implementations
|
|
|
|
|
|
try:
|
|
import pandas
|
|
except ImportError:
|
|
pandas = False # type: ignore
|
|
|
|
try:
|
|
import polars
|
|
except ImportError:
|
|
polars = False # type: ignore
|
|
try:
|
|
import numpy as np
|
|
booleanTypes: Tuple[type, ...] = (bool, np.bool_) # type: ignore
|
|
except ImportError:
|
|
booleanTypes = (bool,) # type: ignore
|
|
|
|
logger: logging.Logger = logging.getLogger(__name__)
|
|
|
|
UNPROCESSED_KEY: object = object()
|
|
|
|
EMPTY_FROZENSET: frozenset = frozenset()
|
|
|
|
INDEX_VS_ATTRIBUTE: Tuple[str, str] = ('[%s]', '.%s')
|
|
|
|
|
|
HASH_LOOKUP_ERR_MSG: str = '{} is not one of the hashed items.'
|
|
|
|
|
|
def sha256hex(obj: Union[str, bytes]) -> str:
|
|
"""Use Sha256 as a cryptographic hash."""
|
|
if isinstance(obj, str):
|
|
obj = obj.encode('utf-8')
|
|
return sha256(obj).hexdigest()
|
|
|
|
|
|
def sha1hex(obj: Union[str, bytes]) -> str:
|
|
"""Use Sha1 as a cryptographic hash."""
|
|
if isinstance(obj, str):
|
|
obj = obj.encode('utf-8')
|
|
return sha1(obj).hexdigest()
|
|
|
|
|
|
default_hasher: Callable[[Union[str, bytes]], str] = sha256hex
|
|
|
|
|
|
class HashLevel:
|
|
"""
|
|
Minimal level-like view of a single object, used to ask a custom operator whether
|
|
it applies to a value that is about to be hashed.
|
|
|
|
DeepHash walks one object at a time, so there is no t1/t2 pair to hand to the
|
|
operator's match method. The value is therefore exposed as both t1 and t2, which
|
|
makes an operator scoped by type or by path behave the same way it would when the
|
|
value is compared against itself.
|
|
"""
|
|
__slots__ = ('t1', 't2', 'up', 'down', '_path')
|
|
|
|
def __init__(self, obj: Any, parent: str) -> None:
|
|
self.t1 = obj
|
|
self.t2 = obj
|
|
self.up = None
|
|
self.down = None
|
|
self._path = parent
|
|
|
|
def path(self, *args: Any, **kwargs: Any) -> str:
|
|
return self._path
|
|
|
|
|
|
def combine_hashes_lists(items: List[List[str]], prefix: Union[str, bytes]) -> str:
|
|
"""
|
|
Combines lists of hashes into one hash
|
|
This can be optimized in future.
|
|
It needs to work with both murmur3 hashes (int) and sha256 (str)
|
|
Although murmur3 is not used anymore.
|
|
"""
|
|
if isinstance(prefix, bytes):
|
|
prefix = prefix.decode('utf-8')
|
|
hashes_bytes = b''
|
|
for item in items:
|
|
# In order to make sure the order of hashes in each item does not affect the hash
|
|
# we resort them.
|
|
hashes_bytes += (''.join(map(str, sorted(item))) + '--').encode('utf-8')
|
|
return prefix + str(default_hasher(hashes_bytes))
|
|
|
|
|
|
class BoolObj(Enum):
|
|
TRUE = 1
|
|
FALSE = 0
|
|
|
|
|
|
def prepare_string_for_hashing(
|
|
obj: Union[str, bytes, memoryview],
|
|
ignore_string_type_changes: bool = False,
|
|
ignore_string_case: bool = False,
|
|
encodings: Optional[List[str]] = None,
|
|
ignore_encoding_errors: bool = False,
|
|
) -> str:
|
|
"""
|
|
Clean type conversions
|
|
"""
|
|
original_type = obj.__class__.__name__
|
|
# https://docs.python.org/3/library/codecs.html#codecs.decode
|
|
errors_mode = 'ignore' if ignore_encoding_errors else 'strict'
|
|
if isinstance(obj, memoryview):
|
|
obj = obj.tobytes()
|
|
if isinstance(obj, bytes):
|
|
err = None
|
|
encodings = ['utf-8'] if encodings is None else encodings
|
|
encoded = False
|
|
for encoding in encodings:
|
|
try:
|
|
obj = obj.decode(encoding, errors=errors_mode)
|
|
encoded = True
|
|
break
|
|
except UnicodeDecodeError as er:
|
|
err = er
|
|
if not encoded and err is not None:
|
|
obj_decoded = obj.decode('utf-8', errors='ignore') # type: ignore
|
|
start = max(err.start - 20, 0)
|
|
start_prefix = ''
|
|
if start > 0:
|
|
start_prefix = '...'
|
|
end = err.end + 20
|
|
end_suffix = '...'
|
|
if end >= len(obj):
|
|
end = len(obj)
|
|
end_suffix = ''
|
|
raise UnicodeDecodeError(
|
|
err.encoding,
|
|
err.object,
|
|
err.start,
|
|
err.end,
|
|
f"{err.reason} in '{start_prefix}{obj_decoded[start:end]}{end_suffix}'. Please either pass ignore_encoding_errors=True or pass the encoding via encodings=['utf-8', '...']."
|
|
) from None
|
|
if not ignore_string_type_changes:
|
|
obj = KEY_TO_VAL_STR.format(original_type, obj)
|
|
if ignore_string_case:
|
|
obj = obj.lower()
|
|
return str(obj)
|
|
|
|
|
|
doc = get_doc('deephash_doc.rst')
|
|
|
|
|
|
class DeepHash(Base):
|
|
__doc__ = doc
|
|
|
|
# Class attributes
|
|
hashes: Dict[Any, Any]
|
|
exclude_types_tuple: Tuple[type, ...]
|
|
ignore_repetition: bool
|
|
exclude_paths: Optional[SetOrdered]
|
|
include_paths: Optional[SetOrdered]
|
|
exclude_regex_paths: Optional[List[re.Pattern[str]]]
|
|
hasher: Callable[[Union[str, bytes]], str]
|
|
use_enum_value: bool
|
|
default_timezone: Union[datetime.timezone, "BaseTzInfo"]
|
|
significant_digits: Optional[int]
|
|
truncate_datetime: Optional[str]
|
|
number_format_notation: str
|
|
ignore_type_in_groups: Any
|
|
ignore_string_type_changes: bool
|
|
ignore_numeric_type_changes: bool
|
|
ignore_string_case: bool
|
|
exclude_obj_callback: Optional[Callable[[Any, str], bool]]
|
|
apply_hash: bool
|
|
type_check_func: Callable[[type, Any], bool]
|
|
number_to_string: Any
|
|
ignore_private_variables: bool
|
|
encodings: Optional[List[str]]
|
|
ignore_encoding_errors: bool
|
|
ignore_iterable_order: bool
|
|
custom_operators: Optional[List[Any]]
|
|
|
|
def __init__(self,
|
|
obj: Any,
|
|
*,
|
|
apply_hash: bool = True,
|
|
custom_operators: Optional[List[Any]] = None,
|
|
default_timezone: Union[datetime.timezone, "BaseTzInfo"] = datetime.timezone.utc,
|
|
encodings: Optional[List[str]] = None,
|
|
exclude_glob_paths: Optional[List[Any]] = None,
|
|
exclude_obj_callback: Optional[Callable[[Any, str], bool]] = None,
|
|
exclude_paths: Optional[PathType] = None,
|
|
exclude_regex_paths: Optional[RegexType] = None,
|
|
exclude_types: Optional[Union[List[type], Set[type], Tuple[type, ...]]] = None,
|
|
hasher: Optional[Callable[[Union[str, bytes]], str]] = None,
|
|
hashes: Optional[Union[Dict[Any, Any], "DeepHash"]] = None,
|
|
ignore_encoding_errors: bool = False,
|
|
ignore_iterable_order: bool = True,
|
|
ignore_numeric_type_changes: bool = False,
|
|
ignore_private_variables: bool = True,
|
|
ignore_repetition: bool = True,
|
|
ignore_string_case: bool = False,
|
|
ignore_string_type_changes: bool = False,
|
|
ignore_type_in_groups: Any = None,
|
|
ignore_type_subclasses: bool = False,
|
|
ignore_uuid_types: bool = False,
|
|
include_glob_paths: Optional[List[Any]] = None,
|
|
include_paths: Optional[PathType] = None,
|
|
number_format_notation: str = "f",
|
|
number_to_string_func: Optional[NumberToStringFunc] = None,
|
|
parent: str = "root",
|
|
significant_digits: Optional[int] = None,
|
|
truncate_datetime: Optional[str] = None,
|
|
use_enum_value: bool = False,
|
|
**kwargs) -> None:
|
|
if kwargs:
|
|
raise ValueError(
|
|
("The following parameter(s) are not valid: %s\n"
|
|
"The valid parameters are obj, hashes, exclude_types, significant_digits, truncate_datetime,"
|
|
"exclude_paths, include_paths, exclude_regex_paths, hasher, ignore_repetition, "
|
|
"number_format_notation, apply_hash, ignore_type_in_groups, ignore_string_type_changes, "
|
|
"ignore_numeric_type_changes, ignore_type_subclasses, ignore_string_case, ignore_uuid_types, "
|
|
"number_to_string_func, ignore_private_variables, parent, use_enum_value, default_timezone "
|
|
"encodings, ignore_encoding_errors") % ', '.join(kwargs.keys()))
|
|
if isinstance(hashes, MutableMapping):
|
|
self.hashes = hashes
|
|
elif isinstance(hashes, DeepHash):
|
|
self.hashes = hashes.hashes
|
|
else:
|
|
self.hashes = dict_()
|
|
exclude_types = set() if exclude_types is None else set(exclude_types)
|
|
self.exclude_types_tuple = tuple(exclude_types) # we need tuple for checking isinstance
|
|
self.ignore_repetition = ignore_repetition
|
|
_exclude_set = convert_item_or_items_into_set_else_none(exclude_paths)
|
|
_exclude_exact, _exclude_globs = separate_wildcard_and_exact_paths(_exclude_set)
|
|
self.exclude_paths = add_root_to_paths(_exclude_exact)
|
|
self.exclude_glob_paths = exclude_glob_paths or _exclude_globs
|
|
_include_set = convert_item_or_items_into_set_else_none(include_paths)
|
|
_include_exact, _include_globs = separate_wildcard_and_exact_paths(_include_set)
|
|
self.include_paths = add_root_to_paths(_include_exact)
|
|
self.include_glob_paths = include_glob_paths or _include_globs
|
|
self.exclude_regex_paths = convert_item_or_items_into_compiled_regexes_else_none(exclude_regex_paths)
|
|
self.hasher = default_hasher if hasher is None else hasher
|
|
self.hashes[UNPROCESSED_KEY] = [] # type: ignore
|
|
self.use_enum_value = use_enum_value
|
|
self.default_timezone = default_timezone
|
|
self.significant_digits = self.get_significant_digits(significant_digits, ignore_numeric_type_changes)
|
|
self.truncate_datetime = get_truncate_datetime(truncate_datetime)
|
|
self.number_format_notation = number_format_notation
|
|
self.ignore_type_in_groups = self.get_ignore_types_in_groups(
|
|
ignore_type_in_groups=ignore_type_in_groups,
|
|
ignore_string_type_changes=ignore_string_type_changes,
|
|
ignore_numeric_type_changes=ignore_numeric_type_changes,
|
|
ignore_type_subclasses=ignore_type_subclasses,
|
|
ignore_uuid_types=ignore_uuid_types,
|
|
)
|
|
self.ignore_string_type_changes = ignore_string_type_changes
|
|
self.ignore_numeric_type_changes = ignore_numeric_type_changes
|
|
self.ignore_string_case = ignore_string_case
|
|
self.exclude_obj_callback = exclude_obj_callback
|
|
# makes the hash return constant size result if true
|
|
# the only time it should be set to False is when
|
|
# testing the individual hash functions for different types of objects.
|
|
self.apply_hash = apply_hash
|
|
self.type_check_func = type_in_type_group if ignore_type_subclasses else type_is_subclass_of_type_group
|
|
# self.type_check_func = type_is_subclass_of_type_group if ignore_type_subclasses else type_in_type_group
|
|
self.number_to_string = number_to_string_func or number_to_string
|
|
self.ignore_private_variables = ignore_private_variables
|
|
self.encodings = encodings
|
|
self.ignore_encoding_errors = ignore_encoding_errors
|
|
self.ignore_iterable_order = ignore_iterable_order
|
|
self.custom_operators = custom_operators
|
|
|
|
self._hash(obj, parent=parent, parents_ids=frozenset({get_id(obj)}))
|
|
|
|
if self.hashes[UNPROCESSED_KEY]:
|
|
logger.warning("Can not hash the following items: {}.".format(self.hashes[UNPROCESSED_KEY]))
|
|
else:
|
|
del self.hashes[UNPROCESSED_KEY]
|
|
|
|
sha256hex: Callable[[Union[str, bytes]], str] = sha256hex
|
|
sha1hex: Callable[[Union[str, bytes]], str] = sha1hex
|
|
|
|
def __getitem__(self, obj: Any, extract_index: Optional[int] = 0) -> Any:
|
|
return self._getitem(self.hashes, obj, extract_index=extract_index,
|
|
use_enum_value=self.use_enum_value,
|
|
ignore_numeric_type_changes=self.ignore_numeric_type_changes)
|
|
|
|
@staticmethod
|
|
def _get_slots_dict(obj: Any) -> Dict[str, Any]:
|
|
"""Get a dict of initialized slot attributes.
|
|
|
|
Uses object.__getattribute__ to check each slot directly, bypassing
|
|
__getattr__. For uninitialized slots on classes that define __getattr__,
|
|
falls back to getattr — letting it raise if the object is truly broken.
|
|
"""
|
|
result = {}
|
|
has_getattr = hasattr(type(obj), '__getattr__')
|
|
for slot in obj.__slots__:
|
|
try:
|
|
result[slot] = object.__getattribute__(obj, slot)
|
|
except AttributeError:
|
|
if has_getattr:
|
|
# The slot isn't initialized, but the class defines __getattr__.
|
|
# Try the normal getattr to let __getattr__ provide a value or
|
|
# raise — if it raises, we propagate to fail the strategy.
|
|
result[slot] = getattr(obj, slot)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _getitem(hashes: Dict[Any, Any], obj: Any, extract_index: Optional[int] = 0,
|
|
use_enum_value: bool = False, ignore_numeric_type_changes: bool = False) -> Any:
|
|
"""
|
|
extract_index is zero for hash and 1 for count and None to get them both.
|
|
To keep it backward compatible, we only get the hash by default so it is set to zero by default.
|
|
"""
|
|
|
|
key = obj
|
|
if obj is True:
|
|
key = BoolObj.TRUE
|
|
elif obj is False:
|
|
key = BoolObj.FALSE
|
|
elif use_enum_value and isinstance(obj, Enum):
|
|
key = obj.value
|
|
key = DeepHash._make_hash_key_for_lookup(key, ignore_numeric_type_changes=ignore_numeric_type_changes)
|
|
|
|
result_n_count: Tuple[Any, int] = (None, 0) # type: ignore
|
|
|
|
try:
|
|
result_n_count = hashes[key]
|
|
except (TypeError, KeyError):
|
|
key = get_id(obj)
|
|
try:
|
|
result_n_count = hashes[key]
|
|
except KeyError:
|
|
raise KeyError(HASH_LOOKUP_ERR_MSG.format(obj)) from None
|
|
|
|
if obj is UNPROCESSED_KEY:
|
|
extract_index = None
|
|
|
|
return result_n_count if extract_index is None else result_n_count[extract_index]
|
|
|
|
def __contains__(self, obj: Any) -> bool:
|
|
key = self._make_hash_key(obj)
|
|
result = False
|
|
try:
|
|
result = key in self.hashes
|
|
except (TypeError, KeyError):
|
|
result = False
|
|
if not result:
|
|
result = get_id(obj) in self.hashes
|
|
return result
|
|
|
|
def get(self, key: Any, default: Any = None, extract_index: Optional[int] = 0) -> Any:
|
|
"""
|
|
Get method for the hashes dictionary.
|
|
It can extract the hash for a given key that is already calculated when extract_index=0
|
|
or the count of items that went to building the object when extract_index=1.
|
|
"""
|
|
return self.get_key(self.hashes, key, default=default, extract_index=extract_index,
|
|
ignore_numeric_type_changes=self.ignore_numeric_type_changes)
|
|
|
|
@staticmethod
|
|
def get_key(hashes: Dict[Any, Any], key: Any, default: Any = None, extract_index: Optional[int] = 0,
|
|
use_enum_value: bool = False, ignore_numeric_type_changes: bool = False) -> Any:
|
|
"""
|
|
get_key method for the hashes dictionary.
|
|
It can extract the hash for a given key that is already calculated when extract_index=0
|
|
or the count of items that went to building the object when extract_index=1.
|
|
"""
|
|
try:
|
|
result = DeepHash._getitem(hashes, key, extract_index=extract_index,
|
|
use_enum_value=use_enum_value,
|
|
ignore_numeric_type_changes=ignore_numeric_type_changes)
|
|
except KeyError:
|
|
result = default
|
|
return result
|
|
|
|
@staticmethod
|
|
def _unwrap_hash_key(key: Any) -> Any:
|
|
"""Unwrap a (type, value) hash key back to the original value for public API."""
|
|
if isinstance(key, tuple) and len(key) == 2 and isinstance(key[0], type) and isinstance(key[1], only_numbers):
|
|
return key[1]
|
|
return key
|
|
|
|
def _get_objects_to_hashes_dict(self, extract_index: Optional[int] = 0) -> Dict[Any, Any]:
|
|
"""
|
|
A dictionary containing only the objects to hashes,
|
|
or a dictionary of objects to the count of items that went to build them.
|
|
extract_index=0 for hashes and extract_index=1 for counts.
|
|
"""
|
|
result = dict_()
|
|
for key, value in self.hashes.items():
|
|
key = self._unwrap_hash_key(key)
|
|
if key is UNPROCESSED_KEY:
|
|
result[key] = value
|
|
else:
|
|
result[key] = value[extract_index]
|
|
return result
|
|
|
|
def __eq__(self, other: Any) -> bool:
|
|
if isinstance(other, DeepHash):
|
|
return self.hashes == other.hashes
|
|
else:
|
|
# We only care about the hashes
|
|
return self._get_objects_to_hashes_dict() == other
|
|
|
|
__req__ = __eq__
|
|
|
|
def __repr__(self) -> str:
|
|
"""
|
|
Hide the counts since it will be confusing to see them when they are hidden everywhere else.
|
|
"""
|
|
from deepdiff.summarize import summarize
|
|
return summarize(self._get_objects_to_hashes_dict(extract_index=0), max_length=500)
|
|
|
|
def __str__(self) -> str:
|
|
return str(self._get_objects_to_hashes_dict(extract_index=0))
|
|
|
|
def __bool__(self) -> bool:
|
|
return bool(self.hashes)
|
|
|
|
def keys(self) -> Any:
|
|
return [self._unwrap_hash_key(k) for k in self.hashes.keys()]
|
|
|
|
def values(self) -> Generator[Any, None, None]:
|
|
return (i[0] for i in self.hashes.values()) # Just grab the item and not its count
|
|
|
|
def items(self) -> Generator[Tuple[Any, Any], None, None]:
|
|
return ((self._unwrap_hash_key(i), v[0]) for i, v in self.hashes.items())
|
|
|
|
def _prep_obj(self, obj: Any, parent: str, parents_ids: frozenset = EMPTY_FROZENSET, is_namedtuple: bool = False, is_pydantic_object: bool = False) -> HashTuple:
|
|
"""prepping objects"""
|
|
original_type = type(obj) if not isinstance(obj, type) else obj
|
|
|
|
obj_to_dict_strategies = []
|
|
if is_namedtuple:
|
|
obj_to_dict_strategies.append(lambda o: o._asdict())
|
|
elif is_pydantic_object:
|
|
obj_to_dict_strategies.append(lambda o: {k: v for (k, v) in o.__dict__.items() if v !="model_fields_set"})
|
|
else:
|
|
obj_to_dict_strategies.append(lambda o: o.__dict__)
|
|
|
|
if hasattr(obj, "__slots__"):
|
|
obj_to_dict_strategies.append(lambda o: DeepHash._get_slots_dict(o))
|
|
else:
|
|
import inspect
|
|
obj_to_dict_strategies.append(lambda o: dict(inspect.getmembers(o, lambda m: not inspect.isroutine(m))))
|
|
|
|
for get_dict in obj_to_dict_strategies:
|
|
try:
|
|
d = get_dict(obj)
|
|
break
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
self.hashes[UNPROCESSED_KEY].append(obj) # type: ignore
|
|
return (unprocessed, 0)
|
|
obj = d
|
|
|
|
result, counts = self._prep_dict(obj, parent=parent, parents_ids=parents_ids,
|
|
print_as_attribute=True, original_type=original_type)
|
|
result = "nt{}".format(result) if is_namedtuple else "obj{}".format(result)
|
|
return result, counts
|
|
|
|
def _skip_this(self, obj: Any, parent: str) -> bool:
|
|
skip = False
|
|
if self.exclude_paths and parent in self.exclude_paths:
|
|
skip = True
|
|
elif self.exclude_glob_paths and any(gp.match(parent) for gp in self.exclude_glob_paths):
|
|
skip = True
|
|
if (self.include_paths or self.include_glob_paths) and parent != 'root':
|
|
skip = True
|
|
if self.include_paths:
|
|
if parent in self.include_paths:
|
|
skip = False
|
|
else:
|
|
for prefix in self.include_paths:
|
|
if parent.startswith(prefix):
|
|
skip = False
|
|
break
|
|
if skip and self.include_glob_paths:
|
|
for gp in self.include_glob_paths:
|
|
if gp.match_or_is_ancestor(parent):
|
|
skip = False
|
|
break
|
|
elif self.exclude_regex_paths and any(
|
|
[exclude_regex_path.search(parent) for exclude_regex_path in self.exclude_regex_paths]): # type: ignore
|
|
skip = True
|
|
elif self.exclude_types_tuple and isinstance(obj, self.exclude_types_tuple):
|
|
skip = True
|
|
elif self.exclude_obj_callback and self.exclude_obj_callback(obj, parent):
|
|
skip = True
|
|
return skip
|
|
|
|
def _prep_dict(self, obj: Union[Dict[Any, Any], MutableMapping], parent: str, parents_ids: frozenset = EMPTY_FROZENSET, print_as_attribute: bool = False, original_type: Optional[type] = None) -> HashTuple:
|
|
|
|
result = []
|
|
counts = 1
|
|
|
|
key_text = "%s{}".format(INDEX_VS_ATTRIBUTE[print_as_attribute])
|
|
for key, item in obj.items():
|
|
counts += 1
|
|
# ignore private variables
|
|
if self.ignore_private_variables and isinstance(key, str) and key.startswith('__'):
|
|
continue
|
|
key_formatted = "'%s'" % key if not print_as_attribute and isinstance(key, strings) else key
|
|
key_in_report = key_text % (parent, key_formatted)
|
|
|
|
key_hash, _ = self._hash(key, parent=key_in_report, parents_ids=parents_ids)
|
|
if not key_hash:
|
|
continue
|
|
item_id = get_id(item)
|
|
if (parents_ids and item_id in parents_ids) or self._skip_this(item, parent=key_in_report):
|
|
continue
|
|
parents_ids_added = add_to_frozen_set(parents_ids, item_id)
|
|
hashed, count = self._hash(item, parent=key_in_report, parents_ids=parents_ids_added)
|
|
hashed = KEY_TO_VAL_STR.format(key_hash, hashed)
|
|
result.append(hashed)
|
|
counts += count
|
|
|
|
result.sort()
|
|
result = ';'.join(result)
|
|
if print_as_attribute:
|
|
type_ = original_type or type(obj)
|
|
type_str = type_.__name__
|
|
for type_group in self.ignore_type_in_groups:
|
|
if self.type_check_func(type_, type_group):
|
|
type_str = ','.join(map(lambda x: x.__name__, type_group))
|
|
break
|
|
else:
|
|
type_str = 'dict'
|
|
return "{}:{{{}}}".format(type_str, result), counts
|
|
|
|
def _prep_iterable(self, obj: Iterable[Any], parent: str, parents_ids: frozenset = EMPTY_FROZENSET) -> HashTuple:
|
|
|
|
counts = 1
|
|
result = defaultdict(int)
|
|
|
|
for i, item in enumerate(obj):
|
|
new_parent = "{}[{}]".format(parent, i)
|
|
if self._skip_this(item, parent=new_parent):
|
|
continue
|
|
|
|
item_id = get_id(item)
|
|
if parents_ids and item_id in parents_ids:
|
|
continue
|
|
|
|
parents_ids_added = add_to_frozen_set(parents_ids, item_id)
|
|
hashed, count = self._hash(item, parent=new_parent, parents_ids=parents_ids_added)
|
|
# counting repetitions
|
|
result[hashed] += 1
|
|
counts += count
|
|
|
|
if self.ignore_repetition:
|
|
result = list(result.keys())
|
|
else:
|
|
result = [
|
|
'{}|{}'.format(i, v) for i, v in result.items()
|
|
]
|
|
|
|
result = map(str, result) # making sure the result items are string so join command works.
|
|
if self.ignore_iterable_order:
|
|
result = sorted(result)
|
|
result = ','.join(result)
|
|
result = KEY_TO_VAL_STR.format(type(obj).__name__, result)
|
|
|
|
return result, counts
|
|
|
|
def _prep_bool(self, obj: bool) -> BoolObj:
|
|
return BoolObj.TRUE if obj else BoolObj.FALSE
|
|
|
|
|
|
def _prep_path(self, obj: Path) -> str:
|
|
type_ = obj.__class__.__name__
|
|
return KEY_TO_VAL_STR.format(type_, obj)
|
|
|
|
def _prep_number(self, obj: Union[int, float, complex]) -> str:
|
|
type_ = "number" if self.ignore_numeric_type_changes else obj.__class__.__name__
|
|
if self.significant_digits is not None:
|
|
obj = self.number_to_string(obj, significant_digits=self.significant_digits,
|
|
number_format_notation=self.number_format_notation) # type: ignore
|
|
return KEY_TO_VAL_STR.format(type_, obj)
|
|
|
|
def _prep_ipranges(self, obj) -> str:
|
|
type_ = 'iprange'
|
|
obj = str(obj)
|
|
return KEY_TO_VAL_STR.format(type_, obj)
|
|
|
|
def _prep_datetime(self, obj: datetime.datetime) -> str:
|
|
type_ = 'datetime'
|
|
obj = datetime_normalize(self.truncate_datetime, obj, default_timezone=self.default_timezone)
|
|
return KEY_TO_VAL_STR.format(type_, obj)
|
|
|
|
def _prep_date(self, obj: datetime.date) -> str:
|
|
type_ = 'datetime' # yes still datetime but it doesn't need normalization
|
|
return KEY_TO_VAL_STR.format(type_, obj)
|
|
|
|
def _prep_tuple(self, obj: tuple, parent: str, parents_ids: frozenset) -> HashTuple:
|
|
# Checking to see if it has _fields. Which probably means it is a named
|
|
# tuple.
|
|
try:
|
|
obj._asdict # type: ignore
|
|
# It must be a normal tuple
|
|
except AttributeError:
|
|
result, counts = self._prep_iterable(obj=obj, parent=parent, parents_ids=parents_ids)
|
|
# We assume it is a namedtuple then
|
|
else:
|
|
result, counts = self._prep_obj(obj, parent, parents_ids=parents_ids, is_namedtuple=True)
|
|
return result, counts
|
|
|
|
def _make_hash_key(self, obj: Any) -> Any:
|
|
"""
|
|
Create a key for the hashes dict that distinguishes numeric types.
|
|
|
|
In Python, 1 == 1.0 and hash(1) == hash(1.0), so int and float values
|
|
collide as dict keys. When ignore_numeric_type_changes is False, we wrap
|
|
numeric objects as (type, value) tuples so that each type gets its own
|
|
cache entry and its own hash.
|
|
"""
|
|
if not self.ignore_numeric_type_changes and isinstance(obj, only_numbers):
|
|
return (type(obj), obj)
|
|
return obj
|
|
|
|
@staticmethod
|
|
def _make_hash_key_for_lookup(obj: Any, ignore_numeric_type_changes: bool = False) -> Any:
|
|
"""Static version of _make_hash_key for use in static accessor methods."""
|
|
if not ignore_numeric_type_changes and isinstance(obj, only_numbers):
|
|
return (type(obj), obj)
|
|
return obj
|
|
|
|
@staticmethod
|
|
def _operator_matches(operator: Any, obj: Any, parent: str) -> bool:
|
|
"""
|
|
Whether a custom operator is in scope for the value that is about to be hashed.
|
|
|
|
An operator that declines to handle a value when diffing must not be allowed to
|
|
rewrite that same value when hashing it for ignore_order=True, otherwise an
|
|
operator scoped to one type silently normalizes every other type as well.
|
|
|
|
Operators without a match method are considered to apply to everything, which
|
|
keeps older operators working.
|
|
"""
|
|
match = getattr(operator, 'match', None)
|
|
if match is None:
|
|
return True
|
|
return bool(match(HashLevel(obj, parent)))
|
|
|
|
def _hash(self, obj: Any, parent: str, parents_ids: frozenset = EMPTY_FROZENSET) -> HashTuple:
|
|
"""The main hash method"""
|
|
counts = 1
|
|
if self.custom_operators is not None:
|
|
for operator in self.custom_operators:
|
|
if not self._operator_matches(operator, obj, parent):
|
|
continue
|
|
func = getattr(operator, 'normalize_value_for_hashing', None)
|
|
if func is None:
|
|
raise NotImplementedError(f"{operator.__class__.__name__} needs to define a normalize_value_for_hashing method to be compatible with ignore_order=True or iterable_compare_func.".format(operator))
|
|
else:
|
|
obj = func(parent, obj)
|
|
|
|
if isinstance(obj, booleanTypes):
|
|
obj = self._prep_bool(obj)
|
|
result = None
|
|
elif self.use_enum_value and isinstance(obj, Enum):
|
|
obj = obj.value
|
|
else:
|
|
result = not_hashed
|
|
hash_key = self._make_hash_key(obj)
|
|
try:
|
|
result, counts = self.hashes[hash_key]
|
|
except (TypeError, KeyError):
|
|
pass
|
|
else:
|
|
return result, counts
|
|
|
|
if self._skip_this(obj, parent):
|
|
return None, 0
|
|
|
|
elif obj is None:
|
|
result = 'NONE'
|
|
|
|
elif isinstance(obj, strings):
|
|
result = prepare_string_for_hashing(
|
|
obj,
|
|
ignore_string_type_changes=self.ignore_string_type_changes,
|
|
ignore_string_case=self.ignore_string_case,
|
|
encodings=self.encodings,
|
|
ignore_encoding_errors=self.ignore_encoding_errors,
|
|
)
|
|
|
|
elif isinstance(obj, Path):
|
|
result = self._prep_path(obj)
|
|
|
|
elif isinstance(obj, times):
|
|
result = self._prep_datetime(obj) # type: ignore
|
|
|
|
elif isinstance(obj, datetime.date):
|
|
result = self._prep_date(obj)
|
|
|
|
elif isinstance(obj, numbers): # type: ignore
|
|
result = self._prep_number(obj)
|
|
|
|
elif isinstance(obj, ipranges):
|
|
result = self._prep_ipranges(obj)
|
|
|
|
elif isinstance(obj, uuid.UUID):
|
|
# Handle UUID objects (including uuid6.UUID) by using their integer value
|
|
result = str(obj.int)
|
|
|
|
elif isinstance(obj, MutableMapping):
|
|
result, counts = self._prep_dict(obj=obj, parent=parent, parents_ids=parents_ids)
|
|
|
|
elif isinstance(obj, tuple):
|
|
result, counts = self._prep_tuple(obj=obj, parent=parent, parents_ids=parents_ids)
|
|
|
|
elif (pandas and isinstance(obj, pandas.DataFrame)): # type: ignore
|
|
def gen(): # type: ignore
|
|
yield ('dtype', obj.dtypes) # type: ignore
|
|
yield ('index', obj.index) # type: ignore
|
|
yield from obj.items() # type: ignore # which contains (column name, series tuples)
|
|
result, counts = self._prep_iterable(obj=gen(), parent=parent, parents_ids=parents_ids)
|
|
elif (polars and isinstance(obj, polars.DataFrame)): # type: ignore
|
|
def gen():
|
|
yield from obj.columns # type: ignore
|
|
yield from list(obj.schema.items()) # type: ignore
|
|
yield from obj.rows() # type: ignore
|
|
result, counts = self._prep_iterable(obj=gen(), parent=parent, parents_ids=parents_ids)
|
|
|
|
elif isinstance(obj, Iterable):
|
|
result, counts = self._prep_iterable(obj=obj, parent=parent, parents_ids=parents_ids)
|
|
|
|
elif obj == BoolObj.TRUE or obj == BoolObj.FALSE:
|
|
result = 'bool:true' if obj is BoolObj.TRUE else 'bool:false'
|
|
elif isinstance(obj, PydanticBaseModel):
|
|
result, counts = self._prep_obj(obj=obj, parent=parent, parents_ids=parents_ids, is_pydantic_object=True)
|
|
else:
|
|
result, counts = self._prep_obj(obj=obj, parent=parent, parents_ids=parents_ids)
|
|
|
|
if result is not_hashed: # pragma: no cover
|
|
self.hashes[UNPROCESSED_KEY].append(obj) # type: ignore
|
|
|
|
elif result is unprocessed:
|
|
pass
|
|
|
|
elif self.apply_hash:
|
|
if isinstance(obj, strings):
|
|
result_cleaned = result
|
|
else:
|
|
result_cleaned = prepare_string_for_hashing(
|
|
str(result), ignore_string_type_changes=self.ignore_string_type_changes,
|
|
ignore_string_case=self.ignore_string_case)
|
|
result = self.hasher(result_cleaned)
|
|
|
|
# It is important to keep the hash of all objects.
|
|
# The hashes will be later used for comparing the objects.
|
|
# Object to hash when possible otherwise ObjectID to hash
|
|
try:
|
|
self.hashes[hash_key] = (result, counts)
|
|
except TypeError:
|
|
obj_id = get_id(obj)
|
|
self.hashes[obj_id] = (result, counts)
|
|
|
|
return result, counts
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
import doctest
|
|
doctest.testmod()
|