Better generic types

Author: dbrattli

pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
     "Programming Language :: Python :: Implementation :: CPython",
     "Topic :: Software Development :: Libraries :: Python Modules",
 ]
-dependencies = ["typing-extensions>=4.1.1,<5"]
+dependencies = ["typing-extensions>=4.15.0,<5"]
 
 [project.urls]
 Homepage = "http://reactivex.io"

reactivex/abc/observable.py

@@ -2,6 +2,8 @@
 from collections.abc import Callable
 from typing import Generic, TypeVar
 
+from typing_extensions import TypeAliasType
+
 from .disposable import DisposableBase
 from .observer import ObserverBase, OnCompleted, OnError, OnNext
 from .scheduler import SchedulerBase
@@ -41,6 +43,10 @@ def subscribe(
         raise NotImplementedError
 
 
-Subscription = Callable[[ObserverBase[_T_out], SchedulerBase | None], DisposableBase]
+Subscription = TypeAliasType(
+    "Subscription",
+    Callable[[ObserverBase[_T_out], SchedulerBase | None], DisposableBase],
+    type_params=(_T_out,),
+)
 
 __all__ = ["ObservableBase", "Subscription"]

reactivex/operators/_flatmap.py

@@ -1,9 +1,9 @@
 from asyncio import Future
-from collections.abc import Callable
 from typing import Any, TypeVar, Union, cast
 
 from reactivex import Observable, from_, from_future
 from reactivex import operators as ops
+from reactivex.internal import curry_flip
 from reactivex.internal.basic import identity
 from reactivex.typing import Mapper, MapperIndexed
 
@@ -38,95 +38,98 @@ def projection(x: _T1, i: int) -> Observable[Any]:
     )
 
 
+@curry_flip
 def flat_map_(
+    source: Observable[_T1],
     mapper: Mapper[_T1, Observable[_T2]] | None = None,
-) -> Callable[[Observable[_T1]], Observable[_T2]]:
-    def flat_map(source: Observable[_T1]) -> Observable[_T2]:
-        """One of the Following:
-        Projects each element of an observable sequence to an observable
-        sequence and merges the resulting observable sequences into one
-        observable sequence.
-
-        Example:
-            >>> flat_map(source)
-
-        Args:
-            source: Source observable to flat map.
-
-        Returns:
-            An operator function that takes a source observable and returns
-            an observable sequence whose elements are the result of invoking
-            the one-to-many transform function on each element of the
-            input sequence .
-        """
-
-        if callable(mapper):
-            ret = _flat_map_internal(source, mapper=mapper)
-        else:
-            ret = _flat_map_internal(source, mapper=lambda _: mapper)
+) -> Observable[_T2]:
+    """Projects each element of an observable sequence to an observable
+    sequence and merges the resulting observable sequences into one
+    observable sequence.
+
+    Examples:
+        >>> source.pipe(flat_map(lambda x: of(x * 2)))
+        >>> flat_map(lambda x: of(x * 2))(source)
+
+    Args:
+        source: Source observable to flat map.
+        mapper: Transform function to apply to each element.
 
-        return ret
+    Returns:
+        An observable sequence whose elements are the result of invoking
+        the one-to-many transform function on each element of the
+        input sequence.
+    """
 
-    return flat_map
+    if callable(mapper):
+        ret = _flat_map_internal(source, mapper=mapper)
+    else:
+        ret = _flat_map_internal(source, mapper=lambda _: mapper)
 
+    return ret
 
+
+@curry_flip
 def flat_map_indexed_(
+    source: Observable[Any],
     mapper_indexed: Any | None = None,
-) -> Callable[[Observable[Any]], Observable[Any]]:
-    def flat_map_indexed(source: Observable[Any]) -> Observable[Any]:
-        """One of the Following:
-        Projects each element of an observable sequence to an observable
-        sequence and merges the resulting observable sequences into one
-        observable sequence.
-
-        Example:
-            >>> flat_map_indexed(source)
-
-        Args:
-            source: Source observable to flat map.
-
-        Returns:
-            An observable sequence whose elements are the result of invoking
-            the one-to-many transform function on each element of the input
-            sequence.
-        """
-
-        if callable(mapper_indexed):
-            ret = _flat_map_internal(source, mapper_indexed=mapper_indexed)
-        else:
-            ret = _flat_map_internal(source, mapper=lambda _: mapper_indexed)
-        return ret
+) -> Observable[Any]:
+    """Projects each element of an observable sequence to an observable
+    sequence and merges the resulting observable sequences into one
+    observable sequence.
 
-    return flat_map_indexed
+    Examples:
+        >>> source.pipe(flat_map_indexed(lambda x, i: of(x * i)))
+        >>> flat_map_indexed(lambda x, i: of(x * i))(source)
 
+    Args:
+        source: Source observable to flat map.
+        mapper_indexed: Transform function with index to apply to each element.
 
+    Returns:
+        An observable sequence whose elements are the result of invoking
+        the one-to-many transform function on each element of the input
+        sequence.
+    """
+
+    if callable(mapper_indexed):
+        ret = _flat_map_internal(source, mapper_indexed=mapper_indexed)
+    else:
+        ret = _flat_map_internal(source, mapper=lambda _: mapper_indexed)
+    return ret
+
+
+@curry_flip
 def flat_map_latest_(
+    source: Observable[_T1],
     mapper: Mapper[_T1, Union[Observable[_T2], "Future[_T2]"]],
-) -> Callable[[Observable[_T1]], Observable[_T2]]:
-    def flat_map_latest(source: Observable[_T1]) -> Observable[_T2]:
-        """Projects each element of an observable sequence into a new
-        sequence of observable sequences by incorporating the element's
-        index and then transforms an observable sequence of observable
-        sequences into an observable sequence producing values only
-        from the most recent observable sequence.
-
-        Args:
-            source: Source observable to flat map latest.
-
-        Returns:
-            An observable sequence whose elements are the result of
-            invoking the transform function on each element of source
-            producing an observable of Observable sequences and that at
-            any point in time produces the elements of the most recent
-            inner observable sequence that has been received.
-        """
-
-        return source.pipe(
-            ops.map(mapper),
-            ops.switch_latest(),
-        )
+) -> Observable[_T2]:
+    """Projects each element of an observable sequence into a new
+    sequence of observable sequences by incorporating the element's
+    index and then transforms an observable sequence of observable
+    sequences into an observable sequence producing values only
+    from the most recent observable sequence.
+
+    Examples:
+        >>> source.pipe(flat_map_latest(lambda x: of(x * 2)))
+        >>> flat_map_latest(lambda x: of(x * 2))(source)
+
+    Args:
+        source: Source observable to flat map latest.
+        mapper: Transform function to apply to each element.
+
+    Returns:
+        An observable sequence whose elements are the result of
+        invoking the transform function on each element of source
+        producing an observable of Observable sequences and that at
+        any point in time produces the elements of the most recent
+        inner observable sequence that has been received.
+    """
 
-    return flat_map_latest
+    return source.pipe(
+        ops.map(mapper),
+        ops.switch_latest(),
+    )
 
 
 __all__ = ["flat_map_", "flat_map_latest_", "flat_map_indexed_"]

reactivex/operators/_publish.py

@@ -1,34 +1,35 @@
-from collections.abc import Callable
 from typing import TypeVar
 
 from reactivex import ConnectableObservable, Observable, abc, compose
 from reactivex import operators as ops
+from reactivex.internal import curry_flip
 from reactivex.subject import Subject
 from reactivex.typing import Mapper
 
 _TSource = TypeVar("_TSource")
 _TResult = TypeVar("_TResult")
 
 
+@curry_flip
 def publish_(
+    source: Observable[_TSource],
     mapper: Mapper[Observable[_TSource], Observable[_TResult]] | None = None,
-) -> Callable[
-    [Observable[_TSource]], Observable[_TResult] | ConnectableObservable[_TSource]
-]:
+) -> Observable[_TResult] | ConnectableObservable[_TSource]:
     """Returns an observable sequence that is the result of invoking the
     mapper on a connectable observable sequence that shares a single
     subscription to the underlying sequence. This operator is a
     specialization of Multicast using a regular Subject.
 
-    Example:
-        >>> res = publish()
-        >>> res = publish(lambda x: x)
+    Examples:
+        >>> source.pipe(publish())
+        >>> source.pipe(publish(lambda x: x))
+        >>> publish()(source)
 
-    mapper: [Optional] Selector function which can use the
-        multicasted source sequence as many times as needed, without causing
-        multiple subscriptions to the source sequence. Subscribers to the
-        given source will receive all notifications of the source from the
-        time of the subscription on.
+    Args:
+        source: Source observable to publish.
+        mapper: [Optional] Selector function which can use the
+            multicasted source sequence as many times as needed, without
+            causing multiple subscriptions to the source sequence.
 
     Returns:
         An observable sequence that contains the elements of a sequence
@@ -41,14 +42,15 @@ def publish_(
         def factory(scheduler: abc.SchedulerBase | None = None) -> Subject[_TSource]:
             return Subject()
 
-        return ops.multicast(subject_factory=factory, mapper=mapper)
+        return source.pipe(ops.multicast(subject_factory=factory, mapper=mapper))
 
     subject: Subject[_TSource] = Subject()
-    return ops.multicast(subject=subject)
+    return source.pipe(ops.multicast(subject=subject))
 
 
-def share_() -> Callable[[Observable[_TSource]], Observable[_TSource]]:
-    """Share a single subscription among multple observers.
+@curry_flip
+def share_(source: Observable[_TSource]) -> Observable[_TSource]:
+    """Share a single subscription among multiple observers.
 
     Returns a new Observable that multicasts (shares) the original
     Observable. As long as there is at least one Subscriber this
@@ -57,10 +59,22 @@ def share_() -> Callable[[Observable[_TSource]], Observable[_TSource]]:
     Observable.
 
     This is an alias for a composed publish() and ref_count().
+
+    Examples:
+        >>> source.pipe(share())
+        >>> share()(source)
+
+    Args:
+        source: Source observable to share.
+
+    Returns:
+        An observable that shares a single subscription.
     """
-    return compose(
-        ops.publish(),
-        ops.ref_count(),
+    return source.pipe(
+        compose(
+            ops.publish(),
+            ops.ref_count(),
+        )
     )
 
 

reactivex/operators/_sample.py

@@ -13,7 +13,7 @@ def sample_observable(
 ) -> Observable[_T]:
     def subscribe(
         observer: abc.ObserverBase[_T], scheduler: abc.SchedulerBase | None = None
-    ):
+    ) -> abc.DisposableBase:
         at_end = False
         has_value = False
         value: _T = cast(_T, None)

reactivex/operators/_sequenceequal.py

@@ -45,7 +45,7 @@ def sequence_equal(source: Observable[_T]) -> Observable[bool]:
         def subscribe(
             observer: abc.ObserverBase[bool],
             scheduler: abc.SchedulerBase | None = None,
-        ):
+        ) -> abc.DisposableBase:
             donel = [False]
             doner = [False]
             ql: list[_T] = []

reactivex/operators/_singleordefault.py

@@ -16,7 +16,7 @@ def single_or_default_async(source: Observable[_T]) -> Observable[_T]:
         def subscribe(
             observer: abc.ObserverBase[_T],
             scheduler: abc.SchedulerBase | None = None,
-        ):
+        ) -> abc.DisposableBase:
             value = cast(_T, default_value)
             seen_value = False
 

reactivex/operators/_skip.py

@@ -32,7 +32,7 @@ def skip_(source: Observable[_T], count: int) -> Observable[_T]:
     def subscribe(
         observer: abc.ObserverBase[_T],
         scheduler: abc.SchedulerBase | None = None,
-    ):
+    ) -> abc.DisposableBase:
         remaining = count
 
         def on_next(value: _T) -> None:

reactivex/operators/_skiplast.py

@@ -33,7 +33,7 @@ def skip_last_(source: Observable[_T], count: int) -> Observable[_T]:
     def subscribe(
         observer: abc.ObserverBase[_T],
         scheduler: abc.SchedulerBase | None = None,
-    ):
+    ) -> abc.DisposableBase:
         q: list[_T] = []
 
         def on_next(value: _T) -> None: