async_rx.observable.rx_first

Complexity

Total Complexity

1

Size/Duplication

Total Lines	18
Duplicated Lines	0 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
A	rx_first()	0	11	1

from ..protocol import Observable
from .rx_take import rx_take

__all__ = ["rx_first"]


def rx_first(observable: Observable) -> Observable:
    """Create an observale which only take the first event and complete.

    Args:
        observable (Observable): observable source

    Returns:
        (Observable): observable instance

    """
    return rx_take(observable=observable, count=1)