fz0x1/HPI
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

core.error: more generic sort_res_by

Browse source
This commit is contained in:
Dima Gerasimov 2020-10-14 21:17:07 +01:00 • committed by karlicoss
parent fa5e181cf8
commit bdfac96352
1 changed files with 21 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
my/core/error.py
View file

@ -44,25 +44,29 @@ def split_errors(l: Iterable[ResT[T, E]], ET: Type[E]) -> Tuple[Iterable[T], Ite
return (values, errors) return (values, errors)
def sort_res_by(items: Iterable[Res[T]], key: Callable[[T], Any]) -> List[Res[T]]: K = TypeVar('K')
def sort_res_by(items: Iterable[Res[T]], key: Callable[[Any], K]) -> List[Res[T]]:
""" """
The general idea is: just alaways carry errors with the entry that precedes them Sort a sequence potentially interleaved with errors/entries on which the key can't be computed.
The general idea is: the error sticks to the non-error entry that follows it
""" """
# TODO ResT object should hold exception class?...
group = [] group = []
groups = [] groups = []
for i in items: for i in items:
if isinstance(i, Exception): k: Optional[K]
try:
k = key(i)
except Exception as e:
k = None
group.append(i) group.append(i)
else: if k is not None:
groups.append((i, group)) groups.append((k, group))
group = [] group = []
results: List[Res[T]] = [] results: List[Res[T]] = []
for v, errs in sorted(groups, key=lambda p: key(p[0])): for v, grp in sorted(groups, key=lambda p: p[0]): # type: ignore[return-value, arg-type] # TODO SupportsLessThan??
results.extend(errs) results.extend(grp)
results.append(v) results.extend(group) # handle last group (it will always be errors only)
results.extend(group)
return results return results
@ -77,15 +81,15 @@ def test_sort_res_by() -> None:
Exc('second'), Exc('second'),
5, 5,
3, 3,
Exc('xxx'), 'bad',
2, 2,
1, 1,
Exc('last'), Exc('last'),
] ]
results = sort_res_by(ress, lambda x: x) # type: ignore results = sort_res_by(ress, lambda x: int(x)) # type: ignore
assert results == [ assert results == [
1, 1,
Exc('xxx'), 'bad',
2, 2,
3, 3,
Exc('first'), Exc('first'),
@ -94,9 +98,11 @@ def test_sort_res_by() -> None:
Exc('last'), Exc('last'),
] ]
results2 = sort_res_by(ress + [0], lambda x: x) # type: ignore results2 = sort_res_by(ress + [0], lambda x: int(x)) # type: ignore
assert results2 == [Exc('last'), 0] + results[:-1] assert results2 == [Exc('last'), 0] + results[:-1]
assert sort_res_by(['caba', 'a', 'aba', 'daba'], key=lambda x: len(x)) == ['a', 'aba', 'caba', 'daba']
assert sort_res_by([], key=lambda x: x) == [] # type: ignore
# helpers to associate timestamps with the errors (so something meaningful could be displayed on the plots, for example) # helpers to associate timestamps with the errors (so something meaningful could be displayed on the plots, for example)