#!/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 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 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: 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()