Use curry-flipping operators

Author: dbrattli

reactivex/internal/curry.py

@@ -7,7 +7,7 @@
 
 import functools
 from collections.abc import Callable
-from typing import Any, Concatenate, TypeVar
+from typing import Concatenate, TypeVar
 
 from typing_extensions import ParamSpec
 
@@ -16,28 +16,6 @@
 _B = TypeVar("_B")
 
 
-def _curry(
-    args: tuple[Any, ...], arity: int, fun: Callable[..., Any]
-) -> Callable[..., Any]:
-    """Internal curry implementation.
-
-    Args:
-        args: Accumulated arguments
-        arity: Number of arguments remaining to curry
-        fun: The function to curry
-
-    Returns:
-        A curried function wrapper
-    """
-
-    def wrapper(*args_: Any, **kw: Any) -> Any:
-        if arity == 1:
-            return fun(*args, *args_, **kw)
-        return _curry(args + args_, arity - 1, fun)
-
-    return wrapper
-
-
 def curry_flip(
     fun: Callable[Concatenate[_A, _P], _B],
 ) -> Callable[_P, Callable[[_A], _B]]:

reactivex/operators/__init__.py

@@ -1810,9 +1810,9 @@ def materialize() -> Callable[[Observable[_T]], Observable[Notification[_T]]]:
         returns an observable sequence containing the materialized
         notification values from the source sequence.
     """
-    from ._materialize import materialize
+    from ._materialize import materialize_
 
-    return materialize()
+    return materialize_()
 
 
 def max(

reactivex/operators/_asobservable.py

@@ -1,32 +1,34 @@
-from collections.abc import Callable
 from typing import TypeVar
 
 from reactivex import Observable, abc
+from reactivex.internal import curry_flip
 
 _T = TypeVar("_T")
 
 
-def as_observable_() -> Callable[[Observable[_T]], Observable[_T]]:
-    def as_observable(source: Observable[_T]) -> Observable[_T]:
-        """Hides the identity of an observable sequence.
+@curry_flip
+def as_observable_(source: Observable[_T]) -> Observable[_T]:
+    """Hides the identity of an observable sequence.
 
-        Args:
-            source: Observable source to hide identity from.
+    Examples:
+        >>> res = source.pipe(as_observable())
+        >>> res = as_observable()(source)
 
-        Returns:
-            An observable sequence that hides the identity of the
-            source sequence.
-        """
+    Args:
+        source: Observable source to hide identity from.
 
-        def subscribe(
-            observer: abc.ObserverBase[_T],
-            scheduler: abc.SchedulerBase | None = None,
-        ) -> abc.DisposableBase:
-            return source.subscribe(observer, scheduler=scheduler)
+    Returns:
+        An observable sequence that hides the identity of the
+        source sequence.
+    """
 
-        return Observable(subscribe)
+    def subscribe(
+        observer: abc.ObserverBase[_T],
+        scheduler: abc.SchedulerBase | None = None,
+    ) -> abc.DisposableBase:
+        return source.subscribe(observer, scheduler=scheduler)
 
-    return as_observable
+    return Observable(subscribe)
 
 
 __all__ = ["as_observable_"]

reactivex/operators/_buffer.py

@@ -1,81 +1,126 @@
 from collections.abc import Callable
 from typing import Any, TypeVar
 
-from reactivex import Observable, compose
+from reactivex import Observable
 from reactivex import operators as ops
+from reactivex.internal import curry_flip
 
 _T = TypeVar("_T")
 
 
+@curry_flip
 def buffer_(
+    source: Observable[_T],
     boundaries: Observable[Any],
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
-    return compose(
+) -> Observable[list[_T]]:
+    """Buffers elements from the source based on boundary notifications.
+
+    Examples:
+        >>> res = source.pipe(buffer(boundaries))
+        >>> res = buffer(boundaries)(source)
+
+    Args:
+        source: Source observable to buffer.
+        boundaries: Observable that triggers buffer emissions.
+
+    Returns:
+        Observable of lists of buffered elements.
+    """
+    return source.pipe(
         ops.window(boundaries),
         ops.flat_map(ops.to_list()),
     )
 
 
+@curry_flip
 def buffer_when_(
+    source: Observable[_T],
     closing_mapper: Callable[[], Observable[Any]],
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
-    return compose(
+) -> Observable[list[_T]]:
+    """Buffers elements using a closing mapper function.
+
+    Examples:
+        >>> res = source.pipe(buffer_when(lambda: timer(1.0)))
+        >>> res = buffer_when(lambda: timer(1.0))(source)
+
+    Args:
+        source: Source observable to buffer.
+        closing_mapper: Function that returns an observable signaling buffer close.
+
+    Returns:
+        Observable of lists of buffered elements.
+    """
+    return source.pipe(
         ops.window_when(closing_mapper),
         ops.flat_map(ops.to_list()),
     )
 
 
+@curry_flip
 def buffer_toggle_(
-    openings: Observable[Any], closing_mapper: Callable[[Any], Observable[Any]]
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
-    return compose(
+    source: Observable[_T],
+    openings: Observable[Any],
+    closing_mapper: Callable[[Any], Observable[Any]],
+) -> Observable[list[_T]]:
+    """Buffers elements using opening/closing observables.
+
+    Examples:
+        >>> res = source.pipe(buffer_toggle(opens, lambda x: timer(x)))
+        >>> res = buffer_toggle(opens, lambda x: timer(x))(source)
+
+    Args:
+        source: Source observable to buffer.
+        openings: Observable that triggers buffer opening.
+        closing_mapper: Function to create closing observable.
+
+    Returns:
+        Observable of lists of buffered elements.
+    """
+    return source.pipe(
         ops.window_toggle(openings, closing_mapper),
         ops.flat_map(ops.to_list()),
     )
 
 
+@curry_flip
 def buffer_with_count_(
-    count: int, skip: int | None = None
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
+    source: Observable[_T],
+    count: int,
+    skip: int | None = None,
+) -> Observable[list[_T]]:
     """Projects each element of an observable sequence into zero or more
     buffers which are produced based on element count information.
 
     Examples:
-        >>> res = buffer_with_count(10)(xs)
-        >>> res = buffer_with_count(10, 1)(xs)
+        >>> res = source.pipe(buffer_with_count(10))
+        >>> res = buffer_with_count(10)(source)
+        >>> res = source.pipe(buffer_with_count(10, 1))
 
     Args:
+        source: Source observable to buffer.
         count: Length of each buffer.
         skip: [Optional] Number of elements to skip between
             creation of consecutive buffers. If not provided, defaults to
             the count.
 
     Returns:
-        A function that takes an observable source and returns an
-        observable sequence of buffers.
+        An observable sequence of buffers.
     """
+    skip_ = skip if skip is not None else count
 
-    def buffer_with_count(source: Observable[_T]) -> Observable[list[_T]]:
-        nonlocal skip
-
-        if skip is None:
-            skip = count
-
-        def mapper(value: Observable[_T]) -> Observable[list[_T]]:
-            return value.pipe(
-                ops.to_list(),
-            )
-
-        def predicate(value: list[_T]) -> bool:
-            return len(value) > 0
-
-        return source.pipe(
-            ops.window_with_count(count, skip),
-            ops.flat_map(mapper),
-            ops.filter(predicate),
+    def mapper(value: Observable[_T]) -> Observable[list[_T]]:
+        return value.pipe(
+            ops.to_list(),
         )
 
-    return buffer_with_count
+    def predicate(value: list[_T]) -> bool:
+        return len(value) > 0
+
+    return source.pipe(
+        ops.window_with_count(count, skip_),
+        ops.flat_map(mapper),
+        ops.filter(predicate),
+    )
 
 
 __all__ = ["buffer_", "buffer_with_count_", "buffer_when_", "buffer_toggle_"]

reactivex/operators/_bufferwithtime.py

@@ -1,23 +1,43 @@
-from collections.abc import Callable
 from typing import TypeVar
 
 from reactivex import Observable, abc, compose, typing
 from reactivex import operators as ops
+from reactivex.internal import curry_flip
 
 _T = TypeVar("_T")
 
 
+@curry_flip
 def buffer_with_time_(
+    source: Observable[_T],
     timespan: typing.RelativeTime,
     timeshift: typing.RelativeTime | None = None,
     scheduler: abc.SchedulerBase | None = None,
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
+) -> Observable[list[_T]]:
+    """Buffers elements based on timing information.
+
+    Examples:
+        >>> source.pipe(buffer_with_time(1.0))
+        >>> source.pipe(buffer_with_time(1.0, 0.5))
+        >>> buffer_with_time(1.0)(source)
+
+    Args:
+        source: Source observable to buffer.
+        timespan: Length of each buffer.
+        timeshift: Interval between creation of consecutive buffers.
+        scheduler: Scheduler to use for timing.
+
+    Returns:
+        An observable sequence of buffers.
+    """
     if not timeshift:
         timeshift = timespan
 
-    return compose(
-        ops.window_with_time(timespan, timeshift, scheduler),
-        ops.flat_map(ops.to_list()),
+    return source.pipe(
+        compose(
+            ops.window_with_time(timespan, timeshift, scheduler),
+            ops.flat_map(ops.to_list()),
+        )
     )
 
 

reactivex/operators/_bufferwithtimeorcount.py

@@ -1,20 +1,39 @@
-from collections.abc import Callable
 from typing import TypeVar
 
 from reactivex import Observable, abc, compose, typing
 from reactivex import operators as ops
+from reactivex.internal import curry_flip
 
 _T = TypeVar("_T")
 
 
+@curry_flip
 def buffer_with_time_or_count_(
+    source: Observable[_T],
     timespan: typing.RelativeTime,
     count: int,
     scheduler: abc.SchedulerBase | None = None,
-) -> Callable[[Observable[_T]], Observable[list[_T]]]:
-    return compose(
-        ops.window_with_time_or_count(timespan, count, scheduler),
-        ops.flat_map(ops.to_iterable()),
+) -> Observable[list[_T]]:
+    """Buffers elements based on timing and count information.
+
+    Examples:
+        >>> source.pipe(buffer_with_time_or_count(1.0, 10))
+        >>> buffer_with_time_or_count(1.0, 10)(source)
+
+    Args:
+        source: Source observable to buffer.
+        timespan: Maximum time length of each buffer.
+        count: Maximum element count of each buffer.
+        scheduler: Scheduler to use for timing.
+
+    Returns:
+        An observable sequence of buffers.
+    """
+    return source.pipe(
+        compose(
+            ops.window_with_time_or_count(timespan, count, scheduler),
+            ops.flat_map(ops.to_iterable()),
+        )
     )