Moar operators

Author: dbrattli

reactivex/observable/mixins/combination.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable, Iterable
 from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast
 
 if TYPE_CHECKING:
@@ -11,6 +12,7 @@
 
 
 _T = TypeVar("_T", covariant=True)
+_T2 = TypeVar("_T2")
 
 
 class CombinationMixin(Generic[_T]):
@@ -311,3 +313,171 @@ def amb(self, right_source: Observable[_T]) -> Observable[_T]:
         from reactivex import operators as ops
 
         return self._as_observable().pipe(ops.amb(right_source))
+
+    def merge_all(self) -> Observable[Any]:
+        """Merge all inner observables.
+
+        Merges an observable sequence of observable sequences into an observable
+        sequence.
+
+        Examples:
+            Fluent style:
+            >>> result = source_of_sources.merge_all()
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source_of_sources.pipe(ops.merge_all())
+
+        Returns:
+            An observable sequence that merges the elements of the inner sequences.
+
+        See Also:
+            - :func:`merge_all <reactivex.operators.merge_all>`
+            - :meth:`merge`
+            - :meth:`concat_all`
+        """
+        # Cast is safe: merge_all is meant to be called on Observable of Observables.
+        # The fluent API allows chaining this on nested observable sequences where
+        # _T is Observable[_T2]. We return Observable[Any] as the inner type cannot
+        # be statically inferred from _T without higher-kinded types.
+        from reactivex import operators as ops
+
+        op: Callable[[Observable[Any]], Observable[Any]] = cast(
+            "Callable[[Observable[Any]], Observable[Any]]", ops.merge_all()
+        )
+        return self._as_observable().pipe(op)
+
+    def zip_with_iterable(
+        self, second: Iterable[_T2]
+    ) -> Observable[tuple[_T, _T2]]:
+        """Zip with iterable.
+
+        Merges the specified observable sequence and iterable into one observable
+        sequence by creating a tuple whenever both sequences have produced an element
+        at a corresponding index.
+
+        Examples:
+            Fluent style:
+            >>> result = source.zip_with_iterable([1, 2, 3])
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.zip_with_iterable([1, 2, 3]))
+
+        Args:
+            second: Iterable to zip with the source observable.
+
+        Returns:
+            An observable sequence containing the result of combining elements of the
+            sources as a tuple.
+
+        See Also:
+            - :func:`zip_with_iterable <reactivex.operators.zip_with_iterable>`
+            - :meth:`zip`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(ops.zip_with_iterable(second))
+
+    def join(
+        self,
+        right: Observable[_T2],
+        left_duration_mapper: Callable[[_T], Observable[Any]],
+        right_duration_mapper: Callable[[_T2], Observable[Any]],
+    ) -> Observable[tuple[_T, _T2]]:
+        """Join based on overlapping durations.
+
+        Correlates the elements of two sequences based on overlapping durations.
+
+        Examples:
+            Fluent style:
+            >>> result = left.join(
+            ...     right,
+            ...     lambda x: rx.timer(0.5),
+            ...     lambda x: rx.timer(0.5)
+            ... )
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = left.pipe(
+            ...     ops.join(
+            ...         right,
+            ...         lambda x: rx.timer(0.5),
+            ...         lambda x: rx.timer(0.5)
+            ...     )
+            ... )
+
+        Args:
+            right: The right observable sequence to join elements for.
+            left_duration_mapper: A function to select the duration (expressed as an
+                observable sequence) of each element of the left observable sequence,
+                used to determine overlap.
+            right_duration_mapper: A function to select the duration (expressed as an
+                observable sequence) of each element of the right observable sequence,
+                used to determine overlap.
+
+        Returns:
+            An observable sequence that contains elements combined into a tuple from
+            source elements that have an overlapping duration.
+
+        See Also:
+            - :func:`join <reactivex.operators.join>`
+            - :meth:`group_join`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(
+            ops.join(right, left_duration_mapper, right_duration_mapper)
+        )
+
+    def group_join(
+        self,
+        right: Observable[_T2],
+        left_duration_mapper: Callable[[_T], Observable[Any]],
+        right_duration_mapper: Callable[[_T2], Observable[Any]],
+    ) -> Observable[tuple[_T, Observable[_T2]]]:
+        """Group join based on overlapping durations.
+
+        Correlates the elements of two sequences based on overlapping durations, and
+        groups the results.
+
+        Examples:
+            Fluent style:
+            >>> result = left.group_join(
+            ...     right,
+            ...     lambda x: rx.timer(0.5),
+            ...     lambda x: rx.timer(0.5)
+            ... )
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = left.pipe(
+            ...     ops.group_join(
+            ...         right,
+            ...         lambda x: rx.timer(0.5),
+            ...         lambda x: rx.timer(0.5)
+            ...     )
+            ... )
+
+        Args:
+            right: The right observable sequence to join elements for.
+            left_duration_mapper: A function to select the duration (expressed as an
+                observable sequence) of each element of the left observable sequence,
+                used to determine overlap.
+            right_duration_mapper: A function to select the duration (expressed as an
+                observable sequence) of each element of the right observable sequence,
+                used to determine overlap.
+
+        Returns:
+            An observable sequence that contains elements combined into a tuple from
+            source elements that have an overlapping duration.
+
+        See Also:
+            - :func:`group_join <reactivex.operators.group_join>`
+            - :meth:`join`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(
+            ops.group_join(right, left_duration_mapper, right_duration_mapper)
+        )

reactivex/observable/mixins/conditional.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast, overload
 
 if TYPE_CHECKING:
@@ -66,3 +67,69 @@ def default_if_empty(self, default_value: Any = None) -> Observable[Any]:
         from reactivex import operators as ops
 
         return self._as_observable().pipe(ops.default_if_empty(default_value))
+
+    def find(
+        self, predicate: Callable[[_T, int, Observable[_T]], bool]
+    ) -> Observable[_T | None]:
+        """Find first element matching predicate.
+
+        Searches for an element that matches the conditions defined by the
+        specified predicate, and returns the first occurrence within the entire
+        Observable sequence.
+
+        Examples:
+            Fluent style:
+            >>> result = source.find(lambda x, i, obs: x > 3)
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.find(lambda x, i, obs: x > 3))
+
+        Args:
+            predicate: The predicate that defines the conditions of the element to
+                search for. Takes (value, index, source).
+
+        Returns:
+            An observable sequence with the first element that matches the conditions
+            defined by the specified predicate, if found; otherwise, None.
+
+        See Also:
+            - :func:`find <reactivex.operators.find>`
+            - :meth:`find_index`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(ops.find(predicate))
+
+    def find_index(
+        self, predicate: Callable[[_T, int, Observable[_T]], bool]
+    ) -> Observable[int | None]:
+        """Find index of first element matching predicate.
+
+        Searches for an element that matches the conditions defined by the specified
+        predicate, and returns an Observable sequence with the zero-based index of
+        the first occurrence within the entire Observable sequence.
+
+        Examples:
+            Fluent style:
+            >>> result = source.find_index(lambda x, i, obs: x > 3)
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.find_index(lambda x, i, obs: x > 3))
+
+        Args:
+            predicate: The predicate that defines the conditions of the element to
+                search for. Takes (value, index, source).
+
+        Returns:
+            An observable sequence with the zero-based index of the first element
+            that matches the conditions; if not found, returns -1.
+
+        See Also:
+            - :func:`find_index <reactivex.operators.find_index>`
+            - :meth:`find`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(ops.find_index(predicate))

reactivex/observable/mixins/filtering.py

@@ -593,3 +593,121 @@ def skip_while_indexed(
         from reactivex import operators as ops
 
         return self._as_observable().pipe(ops.skip_while_indexed(predicate_indexed))
+
+    def single(self, predicate: typing.Predicate[_T] | None = None) -> Observable[_T]:
+        """Return single element matching predicate.
+
+        Returns the only element of an observable sequence that satisfies the condition
+        in the optional predicate, and reports an exception if there is not exactly one
+        element in the observable sequence.
+
+        Examples:
+            Fluent style:
+            >>> result = source.single()
+            >>> result = source.single(lambda x: x == 42)
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.single())
+            >>> result = source.pipe(ops.single(lambda x: x == 42))
+
+        Args:
+            predicate: A predicate function to evaluate for elements in the source
+                sequence.
+
+        Returns:
+            An observable sequence containing the single element in the observable
+            sequence that satisfies the condition in the predicate.
+
+        Raises:
+            Exception: If there is not exactly one element matching the predicate.
+
+        See Also:
+            - :func:`single <reactivex.operators.single>`
+            - :meth:`single_or_default`
+            - :meth:`first`
+            - :meth:`last`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(ops.single(predicate))
+
+    def single_or_default(
+        self, predicate: typing.Predicate[_T] | None = None, default_value: Any = None
+    ) -> Observable[_T]:
+        """Return single element or default.
+
+        Returns the only element of an observable sequence that matches the predicate,
+        or a default value if no such element exists.
+
+        Examples:
+            Fluent style:
+            >>> result = source.single_or_default()
+            >>> result = source.single_or_default(lambda x: x == 42, 0)
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.single_or_default())
+            >>> result = source.pipe(ops.single_or_default(lambda x: x == 42, 0))
+
+        Args:
+            predicate: A predicate function to evaluate for elements in the source
+                sequence.
+            default_value: The default value if no element matches or sequence is empty.
+
+        Returns:
+            An observable sequence containing the single element in the observable
+            sequence that satisfies the condition in the predicate, or the default
+            value if no such element exists.
+
+        See Also:
+            - :func:`single_or_default <reactivex.operators.single_or_default>`
+            - :meth:`single`
+            - :meth:`first_or_default`
+            - :meth:`last_or_default`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(
+            ops.single_or_default(predicate, default_value)
+        )
+
+    def element_at_or_default(
+        self, index: int, default_value: _T | None = None
+    ) -> Observable[_T]:
+        """Get element at index or default.
+
+        Returns the element at a specified index in a sequence or a default value if
+        the index is out of range.
+
+        Examples:
+            Fluent style:
+            >>> result = source.element_at_or_default(5)
+            >>> result = source.element_at_or_default(5, 0)
+
+            Equivalent pipe style:
+            >>> from reactivex import operators as ops
+            >>> result = source.pipe(ops.element_at_or_default(5))
+            >>> result = source.pipe(ops.element_at_or_default(5, 0))
+
+        Args:
+            index: The zero-based index of the element to retrieve.
+            default_value: The default value if the index is outside the bounds of
+                the source sequence.
+
+        Returns:
+            An observable sequence that produces the element at the specified position
+            in the source sequence, or a default value if the index is outside the
+            bounds of the source sequence.
+
+        See Also:
+            - :func:`element_at_or_default <reactivex.operators.element_at_or_default>`
+            - :meth:`element_at`
+            - :meth:`first_or_default`
+            - :meth:`last_or_default`
+        """
+        from reactivex import operators as ops
+
+        return self._as_observable().pipe(
+            ops.element_at_or_default(index, default_value)
+        )