async_rx.protocol.rx_collector.rx_collector()

Complexity

Conditions

3

Size

Total Lines	78
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from collections import namedtuple
def rx_collector(initial_value: T) -> Collector:
    """Create an observer collector.

    Args:
        initial_value (T): initial value which determin result type (list, dict, base type)

    Returns:
        (Collector[T]): a collector instance

    """

    _is_finish = False
    _has_error = False
    _error = None

    if isinstance(initial_value, dict):
        _dict = dict(initial_value)

        async def _on_next(item: Any):
            nonlocal _dict

The variable _dict does not seem to be defined.

            (k, v) = item
            _dict[k] = v

        def _get_result() -> Any:
            nonlocal _dict

The variable _dict does not seem to be defined for all execution paths.

            return _dict

    elif isinstance(initial_value, list):
        _list = list(initial_value)

        async def _on_next(item: Any):
            nonlocal _list

The variable _list does not seem to be defined for all execution paths.

            _list.append(item)

        def _get_result() -> Any:
            nonlocal _list

The variable _list does not seem to be defined for all execution paths.

            return _list

    else:
        _value = initial_value

        async def _on_next(item: Any):
            nonlocal _value

The variable _value does not seem to be defined.

            _value = item

        def _get_result() -> Any:
            nonlocal _value
            return _value

    async def _on_completed():
        nonlocal _is_finish

The variable _is_finish does not seem to be defined.

        _is_finish = True

    async def _on_error(err: Any):
        nonlocal _has_error, _error

The variable _error does not seem to be defined.

The variable _has_error does not seem to be defined.

        _error = err
        _has_error = True

    def _get_is_finish():
        nonlocal _is_finish
        return _is_finish

    def _get_has_error():
        nonlocal _has_error
        return _has_error

    def _get_error():
        nonlocal _error
        return _error

    return CollectorDefinition(
        on_next=_on_next,
        on_error=_on_error,
        on_completed=_on_completed,
        result=_get_result,
        is_finish=_get_is_finish,
        has_error=_get_has_error,
        error=_get_error,
    )