Jmaa/personal-data
1
0
Fork 0
Code Actions Packages

Additional types

Browse Source
This commit is contained in:
Jon Michael Aanes 2024-08-25 21:53:18 +02:00
parent 6bdd778028
commit 72804d145f
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
1 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
personal_data/util.py
View File

@ -58,6 +58,7 @@ def csv_str_to_value(
return None return None
return s return s
def csv_safe_value(v: object) -> str: def csv_safe_value(v: object) -> str:
if isinstance(v, urllib.parse.ParseResult): if isinstance(v, urllib.parse.ParseResult):
return v.geturl() return v.geturl()
@ -91,6 +92,8 @@ def deduplicate_by_ignoring_certain_fields(
for idx2, second in enumerate(dicts[idx1 + 1 :], idx1 + 1): for idx2, second in enumerate(dicts[idx1 + 1 :], idx1 + 1):
if equals_without_fields(first, second, deduplicate_ignore_columns): if equals_without_fields(first, second, deduplicate_ignore_columns):
to_remove.add(idx2) to_remove.add(idx2)
del idx2, second
del idx1, first
to_remove = sorted(to_remove) to_remove = sorted(to_remove)
while to_remove: while to_remove:
@ -100,10 +103,10 @@ def deduplicate_by_ignoring_certain_fields(
def deduplicate_dicts( def deduplicate_dicts(
dicts: Sequence[dict], dicts: Sequence[dict[str,typing.Any] | frozendict[str,typing.Any]],
deduplicate_mode: data.DeduplicateMode, deduplicate_mode: data.DeduplicateMode,
deduplicate_ignore_columns: list[str], deduplicate_ignore_columns: list[str],
) -> tuple[Sequence[dict], list[str]]: ) -> tuple[Sequence[dict[str,typing.Any]], list[str]]:
assert isinstance(deduplicate_ignore_columns, list), deduplicate_ignore_columns assert isinstance(deduplicate_ignore_columns, list), deduplicate_ignore_columns
fieldnames = [] fieldnames = []
@ -133,7 +136,7 @@ def deduplicate_dicts(
return dicts, fieldnames return dicts, fieldnames
def normalize_dict(d: dict) -> frozendict: def normalize_dict(d: dict[str,typing.Any]) -> frozendict[str,typing.Any]:
return frozendict( return frozendict(
{k: csv_str_to_value(str(v)) for k, v in d.items() if csv_str_to_value(str(v)) is not None}, {k: csv_str_to_value(str(v)) for k, v in d.items() if csv_str_to_value(str(v)) is not None},
) )