✏️ Editing: _filter.py Read Only

from typing import Callable, Optional, TypeVar

from reactivex import Observable, abc
from reactivex.typing import Predicate, PredicateIndexed

_T = TypeVar("_T")

# pylint: disable=redefined-builtin
def filter_(predicate: Predicate[_T]) -> Callable[[Observable[_T]], Observable[_T]]:
    def filter(source: Observable[_T]) -> Observable[_T]:
        """Partially applied filter operator.

Filters the elements of an observable sequence based on a
        predicate.

Example:
            >>> filter(source)

Args:
            source: Source observable to filter.

Returns:
            A filtered observable sequence.
        """

def subscribe(
            observer: abc.ObserverBase[_T], scheduler: Optional[abc.SchedulerBase]
        ) -> abc.DisposableBase:
            def on_next(value: _T):
                try:
                    should_run = predicate(value)
                except Exception as ex:  # pylint: disable=broad-except
                    observer.on_error(ex)
                    return

if should_run:
                    observer.on_next(value)

return source.subscribe(
                on_next, observer.on_error, observer.on_completed, scheduler=scheduler
            )

return Observable(subscribe)

return filter

def filter_indexed_(
    predicate_indexed: Optional[PredicateIndexed[_T]] = None,
) -> Callable[[Observable[_T]], Observable[_T]]:
    def filter_indexed(source: Observable[_T]) -> Observable[_T]:
        """Partially applied indexed filter operator.

Filters the elements of an observable sequence based on a
        predicate by incorporating the element's index.

Example:
            >>> filter_indexed(source)

Args:
            source: Source observable to filter.

Returns:
            A filtered observable sequence.
        """

def subscribe(
            observer: abc.ObserverBase[_T], scheduler: Optional[abc.SchedulerBase]
        ):
            count = 0

def on_next(value: _T):
                nonlocal count
                should_run = True

if predicate_indexed:
                    try:
                        should_run = predicate_indexed(value, count)
                    except Exception as ex:  # pylint: disable=broad-except
                        observer.on_error(ex)
                        return
                    else:
                        count += 1

if should_run:
                    observer.on_next(value)

return source.subscribe(
                on_next, observer.on_error, observer.on_completed, scheduler=scheduler
            )

return Observable(subscribe)

return filter_indexed

__all__ = ["filter_", "filter_indexed_"]

🔐 Sid Gifari File Manager Pro