Skip to content

delta: fill in missing type annotations #651

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jorenham wants to merge 1 commit into
base: dev
Choose a base branch
from
+49 −31
Open

delta: fill in missing type annotations #651

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 49 additions & 31 deletions deepdiff/delta.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import copy
import logging
from typing import List, Dict, IO, Callable, Set, Union, Optional, Any, cast
from typing import TYPE_CHECKING, IO, Callable, Any, cast
from functools import partial, cmp_to_key
from collections.abc import Mapping
from copy import deepcopy
Expand All @@ -22,7 +22,10 @@
from deepdiff.anyset import AnySet
from deepdiff.summarize import summarize

logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from _typeshed import SupportsWrite

logger: logging.Logger = logging.getLogger(__name__)


VERIFICATION_MSG = 'Expected the old value for {} to be {} but it is {}. Error found on: {}. You may want to set force=True, especially if this delta is created by passing flat_rows_list or flat_dict_list'
Expand All @@ -46,7 +49,7 @@
NUMPY_TO_LIST = 'NUMPY_TO_LIST'
NOT_VALID_NUMPY_TYPE = "{} is not a valid numpy type."

doc = get_doc('delta.rst')
doc: str = get_doc('delta.rst')


class DeltaError(ValueError):
Expand All @@ -67,27 +70,42 @@ class Delta:

__doc__ = doc

_reversed_diff: DeepDiff | Mapping | None
bidirectional: bool
always_include_values: bool
diff: DeepDiff | Mapping
mutate: bool
raise_errors: bool
log_errors: bool
serializer: Callable[..., bytes | Any]
deserializer: Callable[..., Any]
force: bool
fill: Any
get_nested_obj: Callable[..., Any]

post_process_paths_to_convert: dict[str, Any]

def __init__(
self,
diff: Union[DeepDiff, Mapping, str, bytes, None]=None,
delta_path: Optional[str]=None,
delta_file: Optional[IO]=None,
delta_diff: Optional[dict]=None,
flat_dict_list: Optional[List[Dict]]=None,
flat_rows_list: Optional[List[FlatDeltaRow]]=None,
diff: DeepDiff | Mapping | str | bytes | None=None,
delta_path: str | None=None,
delta_file: IO | None=None,
delta_diff: dict | None=None,
flat_dict_list: list[dict] | None=None,
flat_rows_list: list[FlatDeltaRow] | None=None,
deserializer: Callable=pickle_load,
log_errors: bool=True,
mutate: bool=False,
raise_errors: bool=False,
safe_to_import: Optional[Set[str]]=None,
safe_to_import: set[str] | None=None,
serializer: Callable=pickle_dump,
verify_symmetry: Optional[bool]=None,
verify_symmetry: bool | None=None,
bidirectional: bool=False,
always_include_values: bool=False,
iterable_compare_func_was_used: Optional[bool]=None,
iterable_compare_func_was_used: bool | None=None,
force: bool=False,
fill: Any=not_found,
):
) -> None:
# for pickle deserializer:
if hasattr(deserializer, '__code__') and 'safe_to_import' in set(deserializer.__code__.co_varnames):
_deserializer = deserializer
Expand Down Expand Up @@ -166,13 +184,13 @@ def _deserializer(obj, safe_to_import=None):
self.get_nested_obj = _get_nested_obj
self.reset()

def __repr__(self):
def __repr__(self) -> str:
return "<Delta: {}>".format(summarize(self.diff, max_length=100)) # type: ignore[arg-type]

def reset(self):
def reset(self) -> None:
self.post_process_paths_to_convert = dict_()

def __add__(self, other):
def __add__(self, other: object) -> Any:
if isinstance(other, numbers) and self._numpy_paths: # type: ignore
raise DeltaNumpyOperatorOverrideError(DELTA_NUMPY_OPERATOR_OVERRIDE_MSG)
if self.mutate:
Expand Down Expand Up @@ -204,7 +222,7 @@ def __add__(self, other):

__radd__ = __add__

def __rsub__(self, other):
def __rsub__(self, other: object) -> Any:
if self._reversed_diff is None:
self._reversed_diff = self._get_reverse_diff()
self.diff, self._reversed_diff = self._reversed_diff, self.diff
Expand Down Expand Up @@ -252,7 +270,7 @@ def _get_elem_and_compare_to_old_value(
if next_element is None or isinstance(next_element, str):
_forced_old_value = {}
else:
_forced_old_value = []
_forced_old_value = []
else:
_forced_old_value = forced_old_value
if action == GET:
Expand Down Expand Up @@ -347,7 +365,7 @@ def _set_new_value(self, parent, parent_to_obj_elem, parent_to_obj_action,
parent_to_obj_action, elements,
to_type=list, from_type=tuple)
if elem != 0 and self.force and isinstance(obj, list) and len(obj) == 0:
# it must have been a dictionary
# it must have been a dictionary
obj = {}
self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
value=obj, action=parent_to_obj_action)
Expand Down Expand Up @@ -419,7 +437,7 @@ def _sort_key_for_item_added(path_and_value):
elements = _path_to_elements(path_and_value[0])
# Example elements: [(4.3, 'GET'), ('b', 'GETATTR'), ('a3', 'GET')]
# We only care about the values in the elements not how to get the values.
return [i[0] for i in elements]
return [i[0] for i in elements]

@staticmethod
def _sort_comparison(left, right):
Expand Down Expand Up @@ -667,7 +685,7 @@ def _do_iterable_opcodes(self):
# import pytest; pytest.set_trace()
obj = self.get_nested_obj(obj=self, elements=elements)
is_obj_tuple = isinstance(obj, tuple)
for opcode in opcodes:
for opcode in opcodes:
if opcode.tag == 'replace':
# Replace items in list a[i1:i2] with b[j1:j2]
transformed.extend(opcode.new_values)
Expand Down Expand Up @@ -837,7 +855,7 @@ def _do_ignore_order(self):
self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem,
value=new_obj, action=parent_to_obj_action)

def _get_reverse_diff(self):
def _get_reverse_diff(self) -> dict[str, Any]:
if not self.bidirectional:
raise ValueError('Please recreate the delta with bidirectional=True')

Expand All @@ -863,7 +881,7 @@ def _get_reverse_diff(self):
reverse_path = path_info['new_path'] if path_info.get('new_path') else path
r_diff[action][reverse_path] = {
'new_value': path_info['old_value'], 'old_value': path_info['new_value']
}
}
elif action == 'type_changes':
r_diff[action] = {}
for path, path_info in info.items():
Expand Down Expand Up @@ -901,7 +919,7 @@ def _get_reverse_diff(self):
r_diff[action][path].append(new_op_code)
return r_diff

def dump(self, file):
def dump(self, file: "SupportsWrite[Any]") -> None:
"""
Dump into file object
"""
Expand All @@ -916,13 +934,13 @@ def dump(self, file):
else:
file.write(self.dumps())

def dumps(self):
def dumps(self) -> bytes | Any:
"""
Return the serialized representation of the object as a bytes object, instead of writing it to a file.
"""
return self.serializer(self.diff)

def to_dict(self):
def to_dict(self) -> dict[Any, Any]:
return dict(self.diff)

def _flatten_iterable_opcodes(self, _parse_path):
Expand Down Expand Up @@ -968,7 +986,7 @@ def _get_flat_row(action, info, _parse_path, keys_and_funcs, report_type_changes
yield FlatDeltaRow(**row)

@staticmethod
def _from_flat_rows(flat_rows_list: List[FlatDeltaRow]):
def _from_flat_rows(flat_rows_list: list[FlatDeltaRow]):
flat_dict_list = (i._asdict() for i in flat_rows_list)
return Delta._from_flat_dicts(flat_dict_list)

Expand Down Expand Up @@ -1081,7 +1099,7 @@ def _from_flat_dicts(flat_dict_list):

return result

def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaDict]:
def to_flat_dicts(self, include_action_in_path: bool = False, report_type_changes: bool = True) -> list[FlatDeltaDict]:
"""
Returns a flat list of actions that is easily machine readable.

Expand All @@ -1094,7 +1112,7 @@ def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True)
{'path': [2], 'value': 3, 'action': 'iterable_item_added'},
]


**Parameters**

include_action_in_path : Boolean, default=False
Expand Down Expand Up @@ -1135,11 +1153,11 @@ def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True)
attribute_added
attribute_removed
"""
return cast(List[FlatDeltaDict], [
return cast(list[FlatDeltaDict], [
i._asdict() for i in self.to_flat_rows(include_action_in_path=include_action_in_path, report_type_changes=report_type_changes)
])

def to_flat_rows(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaRow]:
def to_flat_rows(self, include_action_in_path: bool = False, report_type_changes: bool = True) -> list[FlatDeltaRow]:
"""
Just like to_flat_dicts but returns FlatDeltaRow Named Tuples
"""
Expand Down