1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from typing import Any, Callable
from reactivex import Observable, compose
from reactivex import operators as ops
def is_empty_() -> Callable[[Observable[Any]], Observable[bool]]:
"""Determines whether an observable sequence is empty.
Returns:
An observable sequence containing a single element
determining whether the source sequence is empty.
"""
def mapper(b: bool) -> bool:
return not b
return compose(
ops.some(),
ops.map(mapper),
)
__all__ = ["is_empty_"]
|
