1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
from typing import Callable, List, Optional, TypeVar, cast
from reactivex import Observable, compose
from reactivex import operators as ops
from reactivex.internal.basic import identity
from reactivex.internal.exceptions import SequenceContainsNoElementsError
from reactivex.typing import Comparer
_T = TypeVar("_T")
def first_only(x: List[_T]) -> _T:
if not x:
raise SequenceContainsNoElementsError()
return x[0]
def min_(
comparer: Optional[Comparer[_T]] = None,
) -> Callable[[Observable[_T]], Observable[_T]]:
"""The `min` operator.
Returns the minimum element in an observable sequence according to
the optional comparer else a default greater than less than check.
Examples:
>>> res = source.min()
>>> res = source.min(lambda x, y: x.value - y.value)
Args:
comparer: [Optional] Comparer used to compare elements.
Returns:
An observable sequence containing a single element
with the minimum element in the source sequence.
"""
return compose(
ops.min_by(cast(Callable[[_T], _T], identity), comparer),
ops.map(first_only),
)
__all__ = ["min_"]
|
