Remove average extra check

Fix typing on timestamp

Author: mat

reactivex/operators/_average.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass
-from typing import Any, Callable, Optional, TypeVar, cast
+from typing import Any, Optional, TypeVar, cast
 
 from reactivex import Observable, operators, typing
+from reactivex.curry import curry_flip
 
 _T = TypeVar("_T")
 
@@ -12,51 +13,47 @@ class AverageValue:
     count: int
 
 
+@curry_flip(1)
 def average_(
+    source: Observable[Any],
     key_mapper: Optional[typing.Mapper[_T, float]] = None,
-) -> Callable[[Observable[_T]], Observable[float]]:
-    def average(source: Observable[Any]) -> Observable[float]:
-        """Partially applied average operator.
+) -> Observable[float]:
+    """Partially applied average operator.
 
-        Computes the average of an observable sequence of values that
-        are in the sequence or obtained by invoking a transform
-        function on each element of the input sequence if present.
+    Computes the average of an observable sequence of values that
+    are in the sequence or obtained by invoking a transform
+    function on each element of the input sequence if present.
 
-        Examples:
-            >>> res = average(source)
+    Examples:
+        >>> res = average(source)
 
-        Args:
-            source: Source observable to average.
+    Args:
+        source: Source observable to average.
 
-        Returns:
-            An observable sequence containing a single element with the
-            average of the sequence of values.
-        """
+    Returns:
+        An observable sequence containing a single element with the
+        average of the sequence of values.
+    """
 
-        key_mapper_: typing.Mapper[_T, float] = key_mapper or (
-            lambda x: float(cast(Any, x))
-        )
+    key_mapper_: typing.Mapper[_T, float] = key_mapper or (
+        lambda x: float(cast(Any, x))
+    )
 
-        def accumulator(prev: AverageValue, cur: float) -> AverageValue:
-            return AverageValue(sum=prev.sum + cur, count=prev.count + 1)
+    def accumulator(prev: AverageValue, cur: float) -> AverageValue:
+        return AverageValue(sum=prev.sum + cur, count=prev.count + 1)
 
-        def mapper(s: AverageValue) -> float:
-            if s.count == 0:
-                raise Exception("The input sequence was empty")
+    def mapper(s: AverageValue) -> float:
+        return s.sum / float(s.count)
 
-            return s.sum / float(s.count)
+    seed = AverageValue(sum=0, count=0)
 
-        seed = AverageValue(sum=0, count=0)
-
-        ret = source.pipe(
-            operators.map(key_mapper_),
-            operators.scan(accumulator, seed),
-            operators.last(),
-            operators.map(mapper),
-        )
-        return ret
-
-    return average
+    ret = source.pipe(
+        operators.map(key_mapper_),
+        operators.scan(accumulator, seed),
+        operators.last(),
+        operators.map(mapper),
+    )
+    return ret
 
 
 __all__ = ["average_"]

reactivex/operators/_take.py

@@ -1,53 +1,52 @@
-from typing import Callable, Optional, TypeVar
+from typing import Optional, TypeVar, cast
 
 from reactivex import Observable, abc, empty
+from reactivex.curry import curry_flip
 from reactivex.internal import ArgumentOutOfRangeException
 
 _T = TypeVar("_T")
 
 
-def take_(count: int) -> Callable[[Observable[_T]], Observable[_T]]:
+@curry_flip(1)
+def take_(source: Observable[_T], count: int) -> Observable[_T]:
     if count < 0:
         raise ArgumentOutOfRangeException()
 
-    def take(source: Observable[_T]) -> Observable[_T]:
-        """Returns a specified number of contiguous elements from the start of
-        an observable sequence.
+    """Returns a specified number of contiguous elements from the start of
+    an observable sequence.
 
-        >>> take(source)
+    >>> take(source)
 
-        Keyword arguments:
-        count -- The number of elements to return.
+    Keyword arguments:
+    count -- The number of elements to return.
 
-        Returns an observable sequence that contains the specified number of
-        elements from the start of the input sequence.
-        """
+    Returns an observable sequence that contains the specified number of
+    elements from the start of the input sequence.
+    """
 
-        if not count:
-            return empty()
+    if not count:
+        return cast(Observable[_T], empty())
 
-        def subscribe(
-            observer: abc.ObserverBase[_T],
-            scheduler: Optional[abc.SchedulerBase] = None,
-        ):
-            remaining = count
+    def subscribe(
+        observer: abc.ObserverBase[_T],
+        scheduler: Optional[abc.SchedulerBase] = None,
+    ):
+        remaining = count
 
-            def on_next(value: _T) -> None:
-                nonlocal remaining
+        def on_next(value: _T) -> None:
+            nonlocal remaining
 
-                if remaining > 0:
-                    remaining -= 1
-                    observer.on_next(value)
-                    if not remaining:
-                        observer.on_completed()
+            if remaining > 0:
+                remaining -= 1
+                observer.on_next(value)
+                if not remaining:
+                    observer.on_completed()
 
-            return source.subscribe(
-                on_next, observer.on_error, observer.on_completed, scheduler=scheduler
-            )
+        return source.subscribe(
+            on_next, observer.on_error, observer.on_completed, scheduler=scheduler
+        )
 
-        return Observable(subscribe)
-
-    return take
+    return Observable(subscribe)
 
 
 __all__ = ["take_"]

reactivex/operators/_throttlefirst.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Callable, Optional, TypeVar
+from typing import Optional, TypeVar
 
 from reactivex import Observable, abc, typing
 from reactivex.curry import curry_flip
@@ -13,7 +13,7 @@ def throttle_first_(
     source: Observable[_T],
     window_duration: typing.RelativeTime,
     scheduler: Optional[abc.SchedulerBase] = None,
-) -> Callable[[Observable[_T]], Observable[_T]]:
+) -> Observable[_T]:
     """Returns an observable that emits only the first item emitted
     by the source Observable during sequential time windows of a
     specified duration.

reactivex/operators/_timestamp.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Callable, Generic, Optional, TypeVar
+from typing import Generic, Optional, TypeVar
 
 from reactivex import Observable, abc, defer, operators
 from reactivex.curry import curry_flip
@@ -19,7 +19,7 @@ class Timestamp(Generic[_T]):
 def timestamp_(
     source: Observable[_T],
     scheduler: Optional[abc.SchedulerBase] = None,
-) -> Callable[[Observable[_T]], Observable[Timestamp[_T]]]:
+) -> Observable[Timestamp[_T]]:
     """Records the timestamp for each value in an observable sequence.
 
     Examples:

tests/test_observable/test_average.py

@@ -1,6 +1,7 @@
 import unittest
 
 from reactivex import operators as _
+from reactivex.internal.exceptions import SequenceContainsNoElementsError
 from reactivex.testing import ReactiveTest, TestScheduler
 
 on_next = ReactiveTest.on_next
@@ -19,9 +20,7 @@ def test_average_int32_empty(self):
         xs = scheduler.create_hot_observable(msgs)
         res = scheduler.start(create=lambda: xs.pipe(_.average())).messages
 
-        assert len(res) == 1
-        assert res[0].value.kind == "E" and res[0].value.exception != None
-        assert res[0].time == 250
+        assert res == [on_error(250, SequenceContainsNoElementsError())]
 
     def test_average_int32_return(self):
         scheduler = TestScheduler()