API Reference

All public symbols are importable from the top-level asyncio_util package:

from asyncio_util import AsyncValue, compose_values, wait_any, ...

For trio-util compatibility, BoolEvent and ValueEvent are provided as aliases of AsyncBool and AsyncValue.

Awaitable values

asyncio_util.AsyncValue

Bases: Generic[T]

A mutable value whose changes can be awaited.

AsyncValue wraps a value and lets any number of tasks wait for states ("the value is 20") or transitions ("the value just changed to 20") without polling and without managing events by hand.

Matching is evaluated synchronously at assignment time, so :meth:wait_value and :meth:wait_transition never miss a match even if the value changes again immediately afterwards.

Assigning a value equal to the current one is a no-op and wakes nobody.

Predicates must be fast, side-effect free, and must not raise: they run synchronously inside the value setter. Because any callable is treated as a predicate, a value that is itself callable cannot be waited on by equality — wrap it: lambda v: v == target. If a predicate does raise, the exception is delivered to that waiter (raised from its await); other waiters are unaffected.

Note

AsyncValue is not thread-safe. Assign value only from the thread running the event loop; from other threads use loop.call_soon_threadsafe.

Example

av = AsyncValue(0)

async def watcher():
    value = await av.wait_value(20)   # suspends until 20
    print(value)

# elsewhere, later:
av.value = 20                          # wakes the watcher

Source code in src/asyncio_util/_async_value.py

class AsyncValue(Generic[T]):
    """A mutable value whose changes can be awaited.

    ``AsyncValue`` wraps a value and lets any number of tasks wait for
    states ("the value is 20") or transitions ("the value just changed
    to 20") without polling and without managing events by hand.

    Matching is evaluated synchronously at assignment time, so
    :meth:`wait_value` and :meth:`wait_transition` never miss a match
    even if the value changes again immediately afterwards.

    Assigning a value equal to the current one is a no-op and wakes
    nobody.

    Predicates must be fast, side-effect free, and must not raise: they
    run synchronously inside the ``value`` setter.  Because any callable
    is treated as a predicate, a value that is itself callable cannot be
    waited on by equality — wrap it: ``lambda v: v == target``.  If a
    predicate does raise, the exception is delivered to that waiter
    (raised from its ``await``); other waiters are unaffected.

    Note:
        ``AsyncValue`` is not thread-safe.  Assign ``value`` only from
        the thread running the event loop; from other threads use
        ``loop.call_soon_threadsafe``.

    Example:
        ```python
        av = AsyncValue(0)

        async def watcher():
            value = await av.wait_value(20)   # suspends until 20
            print(value)

        # elsewhere, later:
        av.value = 20                          # wakes the watcher
        ```
    """

    def __init__(self, value: T):
        self._value = value
        self._level_results = _RefCountedDefaultDict(_Result)
        self._edge_results = _RefCountedDefaultDict(_Result)
        self._transforms = _RefCountedDefaultDict(lambda: AsyncValue(None))
        self._listeners: list[Callable[[T, T], None]] = []

    def __repr__(self):
        return f"{self.__class__.__name__}({self.value})"

    @property
    def value(self) -> T:
        """The current value.

        Assigning a new value synchronously notifies all matching
        waiters.  Assigning a value equal to the current one does
        nothing.
        """
        return self._value

    @value.setter
    def value(self, x: T):
        if self._value != x:
            old = self._value
            new = self._value = x
            # A raising predicate or transform must not prevent the
            # remaining waiters from being notified: predicate errors are
            # delivered to their own waiters, transform errors are
            # re-raised only after every notification went out.
            transform_error: Exception | None = None
            for f, result in self._level_results.items():
                try:
                    matched = f(new)
                except Exception as exc:
                    result.event.unpark_all(exc)
                    continue
                if matched:
                    result.value = new
                    result.event.unpark_all()
            for f, result in self._edge_results.items():
                try:
                    matched = f(new, old)
                except Exception as exc:
                    result.event.unpark_all(exc)
                    continue
                if matched:
                    result.value = (new, old)
                    result.event.unpark_all()
            for f, output in self._transforms.items():
                try:
                    output.value = f(new)
                except Exception as exc:
                    transform_error = transform_error or exc
            for listener in list(self._listeners):
                try:
                    listener(new, old)
                except Exception as exc:
                    transform_error = transform_error or exc
            if transform_error is not None:
                raise transform_error

    async def _wait_predicate(self, result_map, predicate):
        with result_map.open_ref(predicate) as result:
            await result.event.park()
            return result.value

    @contextmanager
    def _subscribe(self, listener: Callable[[T, T], None]) -> Iterator[None]:
        """Invoke ``listener(new_value, old_value)`` on every change.

        Listener exceptions are re-raised to the code assigning
        ``value`` after all other notifications went out.
        """
        self._listeners.append(listener)
        try:
            yield
        finally:
            self._listeners.remove(listener)

    @overload
    async def wait_value(self, value: T, *, held_for=0.0, timeout=None) -> T: ...
    @overload
    async def wait_value(self, predicate: P, *, held_for=0.0, timeout=None) -> T: ...
    async def wait_value(
        self,
        value_or_predicate: T | P,
        *,
        held_for: float = 0.0,
        timeout: float | None = None,
    ) -> T:
        """Wait until the value matches, and return the matched value.

        Returns immediately if the current value already matches.
        Otherwise the match is captured at assignment time, so a
        matching value is returned even if the value changes again
        before this coroutine resumes (the current value may differ by
        then).

        Args:
            value_or_predicate: A plain value to compare with ``==``, or
                a callable ``predicate(value) -> bool``.  Any callable
                is treated as a predicate.
            held_for: If greater than 0, only return once the match has
                held for at least that many seconds.  Each time the
                match is lost, the hold timer restarts.
            timeout: If not None, raise :class:`asyncio.TimeoutError`
                when no match occurred within *timeout* seconds.

        Returns:
            The value that satisfied the match.
        """
        predicate = _ValueWrapper(value_or_predicate)
        while True:
            try:
                if not predicate(self._value):
                    value = await asyncio.wait_for(
                        self._wait_predicate(self._level_results, predicate), timeout
                    )
                else:
                    value = self._value
                    await asyncio.sleep(0)
            except asyncio.TimeoutError:
                timeout_message = "Operation timed out"
                raise asyncio.TimeoutError(timeout_message) from None
            if held_for > 0:
                try:
                    await asyncio.wait_for(
                        self.wait_value(lambda v: not predicate(v)), held_for
                    )
                    # The match was lost before the hold elapsed; wait again.
                    continue
                except asyncio.TimeoutError:
                    # The match was held for the full duration.  Re-check in
                    # case the value changed while the timeout was being
                    # processed.
                    if not predicate(self._value):
                        continue
                    value = self._value
            return value

    @overload
    async def eventual_values(self, value: T, held_for=0.0) -> AsyncIterator[T]:
        yield self._value

    @overload
    async def eventual_values(
        self, predicate: P = _any_value, held_for=0.0
    ) -> AsyncIterator[T]:
        yield self._value

    async def eventual_values(
        self,
        value_or_predicate: T | P = _any_value,
        held_for: float = 0.0,
    ) -> AsyncIterator[T]:
        """Iterate over matching values, always converging to the latest.

        Yields the current value first if it matches, then each
        matching value as it is assigned.  A slow consumer may miss
        intermediate values, but is always caught up with the latest
        matching value — hence "eventual".  Consecutive duplicates are
        suppressed.

        Args:
            value_or_predicate: A plain value to compare with ``==``, or
                a callable ``predicate(value) -> bool``.  Omit it to
                match every value.
            held_for: If greater than 0, only yield a value once it has
                matched continuously for that many seconds.

        Example:
            ```python
            async for state in av.eventual_values():
                redraw(state)
            ```
        """
        predicate = _ValueWrapper(value_or_predicate)
        last_value = self._value
        with self._level_results.open_ref(
            predicate
        ) as result, self._level_results.open_ref(
            lambda v: v != last_value
        ) as not_last_value, self._level_results.open_ref(
            lambda v: not predicate(v)
        ) as not_predicate:
            while True:
                if predicate(self._value):
                    last_value = self._value
                else:
                    await result.event.park()
                    last_value = result.value
                if held_for > 0:
                    try:
                        await asyncio.wait_for(not_predicate.event.park(), held_for)
                        # The match was lost before the hold elapsed; start over.
                        continue
                    except asyncio.TimeoutError:
                        # The match was held for the full duration.
                        if not predicate(self._value):
                            continue
                        last_value = self._value
                yield last_value
                if self._value == last_value:
                    await not_last_value.event.park()

    @overload
    async def wait_transition(self, value: T) -> tuple[T, T]: ...
    @overload
    async def wait_transition(self, predicate: P2 = _any_transition) -> tuple[T, T]: ...
    async def wait_transition(
        self,
        value_or_predicate: T | P2 = _any_transition,
        *,
        timeout: float | None = None,
    ) -> tuple[T, T]:
        """Wait for the next matching change, and return ``(value, old_value)``.

        Unlike :meth:`wait_value`, the current value is never a match:
        this waits for an *edge*, an actual assignment that changes the
        value.

        Args:
            value_or_predicate: A plain value the *new* value must equal,
                or a callable ``predicate(value, old_value) -> bool``.
                Omit it to match any change.
            timeout: If not None, raise :class:`asyncio.TimeoutError`
                when no matching change occurred within *timeout*
                seconds.

        Returns:
            The ``(value, old_value)`` pair captured at assignment time.
        """
        try:
            return await asyncio.wait_for(
                self._wait_predicate(self._edge_results, _ValueWrapper(value_or_predicate)),
                timeout,
            )
        except asyncio.TimeoutError:
            raise asyncio.TimeoutError("Operation timed out") from None

    @overload
    async def transitions(self, value: T) -> AsyncIterator[tuple[T, T]]:
        yield (self._value, self._value)

    @overload
    async def transitions(
        self, predicate: P2 = _any_transition
    ) -> AsyncIterator[tuple[T, T]]:
        yield (self._value, self._value)

    async def transitions(
        self,
        value_or_predicate: T | P2 = _any_transition,
    ) -> AsyncIterator[tuple[T, T]]:
        """Iterate over matching changes as ``(value, old_value)`` pairs.

        Like :meth:`wait_transition` in a loop: only edges are reported,
        never the current value.

        Warning:
            Transitions are not queued.  A change that happens while the
            loop body is still processing the previous one is missed.
            If every edge matters, have the producer push into an
            ``asyncio.Queue`` (or a :class:`~asyncio_util.MulticastQueue`)
            instead.

        Args:
            value_or_predicate: A plain value the *new* value must equal,
                or a callable ``predicate(value, old_value) -> bool``.
                Omit it to match any change.
        """
        predicate = _ValueWrapper(value_or_predicate)
        with self._edge_results.open_ref(predicate) as result:
            while True:
                await result.event.park()
                yield result.value

    @asynccontextmanager
    async def open_transform(
        self, function: Callable[[T], T_OUT]
    ) -> AsyncContextManager[AsyncValue[T_OUT]]:
        """Derive a new :class:`AsyncValue` that tracks ``function(value)``.

        Within the ``async with`` block, the derived value is updated
        synchronously on every assignment to this value.  Updates that
        produce an equal output are deduplicated by the derived value's
        own setter.  Opening the same function object twice shares one
        derived value (reference-counted).

        Args:
            function: A pure, non-raising function applied to each value.
                It runs inside the source's setter; if it raises, the
                exception propagates to the code assigning ``value``
                (after all other waiters have been notified).

        Example:
            ```python
            av = AsyncValue(3)
            async with av.open_transform(lambda x: x * 10) as derived:
                assert derived.value == 30
            ```
        """
        with self._transforms.open_ref(function) as output:
            if output.value is None:
                output.value = function(self.value)
            yield output

value property writable

value: T

The current value.

Assigning a new value synchronously notifies all matching waiters. Assigning a value equal to the current one does nothing.

wait_value async

wait_value(value: T, *, held_for=0.0, timeout=None) -> T

wait_value(predicate: P, *, held_for=0.0, timeout=None) -> T

wait_value(value_or_predicate: T | P, *, held_for: float = 0.0, timeout: float | None = None) -> T

Wait until the value matches, and return the matched value.

Returns immediately if the current value already matches. Otherwise the match is captured at assignment time, so a matching value is returned even if the value changes again before this coroutine resumes (the current value may differ by then).

Parameters:

Name	Type	Description	Default
value_or_predicate	T | P	A plain value to compare with ==, or a callable predicate(value) -> bool. Any callable is treated as a predicate.	required
held_for	float	If greater than 0, only return once the match has held for at least that many seconds. Each time the match is lost, the hold timer restarts.	0.0
timeout	float | None	If not None, raise :class:asyncio.TimeoutError when no match occurred within timeout seconds.	None

Returns:

Type	Description
T	The value that satisfied the match.

Source code in src/asyncio_util/_async_value.py

async def wait_value(
    self,
    value_or_predicate: T | P,
    *,
    held_for: float = 0.0,
    timeout: float | None = None,
) -> T:
    """Wait until the value matches, and return the matched value.

    Returns immediately if the current value already matches.
    Otherwise the match is captured at assignment time, so a
    matching value is returned even if the value changes again
    before this coroutine resumes (the current value may differ by
    then).

    Args:
        value_or_predicate: A plain value to compare with ``==``, or
            a callable ``predicate(value) -> bool``.  Any callable
            is treated as a predicate.
        held_for: If greater than 0, only return once the match has
            held for at least that many seconds.  Each time the
            match is lost, the hold timer restarts.
        timeout: If not None, raise :class:`asyncio.TimeoutError`
            when no match occurred within *timeout* seconds.

    Returns:
        The value that satisfied the match.
    """
    predicate = _ValueWrapper(value_or_predicate)
    while True:
        try:
            if not predicate(self._value):
                value = await asyncio.wait_for(
                    self._wait_predicate(self._level_results, predicate), timeout
                )
            else:
                value = self._value
                await asyncio.sleep(0)
        except asyncio.TimeoutError:
            timeout_message = "Operation timed out"
            raise asyncio.TimeoutError(timeout_message) from None
        if held_for > 0:
            try:
                await asyncio.wait_for(
                    self.wait_value(lambda v: not predicate(v)), held_for
                )
                # The match was lost before the hold elapsed; wait again.
                continue
            except asyncio.TimeoutError:
                # The match was held for the full duration.  Re-check in
                # case the value changed while the timeout was being
                # processed.
                if not predicate(self._value):
                    continue
                value = self._value
        return value

eventual_values async

eventual_values(value: T, held_for=0.0) -> AsyncIterator[T]

eventual_values(predicate: P = _any_value, held_for=0.0) -> AsyncIterator[T]

eventual_values(value_or_predicate: T | P = _any_value, held_for: float = 0.0) -> AsyncIterator[T]

Iterate over matching values, always converging to the latest.

Yields the current value first if it matches, then each matching value as it is assigned. A slow consumer may miss intermediate values, but is always caught up with the latest matching value — hence "eventual". Consecutive duplicates are suppressed.

Parameters:

Name	Type	Description	Default
value_or_predicate	T | P	A plain value to compare with ==, or a callable predicate(value) -> bool. Omit it to match every value.	_any_value
held_for	float	If greater than 0, only yield a value once it has matched continuously for that many seconds.	0.0

Example

async for state in av.eventual_values():
    redraw(state)

Source code in src/asyncio_util/_async_value.py

async def eventual_values(
    self,
    value_or_predicate: T | P = _any_value,
    held_for: float = 0.0,
) -> AsyncIterator[T]:
    """Iterate over matching values, always converging to the latest.

    Yields the current value first if it matches, then each
    matching value as it is assigned.  A slow consumer may miss
    intermediate values, but is always caught up with the latest
    matching value — hence "eventual".  Consecutive duplicates are
    suppressed.

    Args:
        value_or_predicate: A plain value to compare with ``==``, or
            a callable ``predicate(value) -> bool``.  Omit it to
            match every value.
        held_for: If greater than 0, only yield a value once it has
            matched continuously for that many seconds.

    Example:
        ```python
        async for state in av.eventual_values():
            redraw(state)
        ```
    """
    predicate = _ValueWrapper(value_or_predicate)
    last_value = self._value
    with self._level_results.open_ref(
        predicate
    ) as result, self._level_results.open_ref(
        lambda v: v != last_value
    ) as not_last_value, self._level_results.open_ref(
        lambda v: not predicate(v)
    ) as not_predicate:
        while True:
            if predicate(self._value):
                last_value = self._value
            else:
                await result.event.park()
                last_value = result.value
            if held_for > 0:
                try:
                    await asyncio.wait_for(not_predicate.event.park(), held_for)
                    # The match was lost before the hold elapsed; start over.
                    continue
                except asyncio.TimeoutError:
                    # The match was held for the full duration.
                    if not predicate(self._value):
                        continue
                    last_value = self._value
            yield last_value
            if self._value == last_value:
                await not_last_value.event.park()

wait_transition async

wait_transition(value: T) -> tuple[T, T]

wait_transition(predicate: P2 = _any_transition) -> tuple[T, T]

wait_transition(value_or_predicate: T | P2 = _any_transition, *, timeout: float | None = None) -> tuple[T, T]

Wait for the next matching change, and return (value, old_value).

Unlike :meth:wait_value, the current value is never a match: this waits for an edge, an actual assignment that changes the value.

Parameters:

Name	Type	Description	Default
value_or_predicate	T | P2	A plain value the new value must equal, or a callable predicate(value, old_value) -> bool. Omit it to match any change.	_any_transition
timeout	float | None	If not None, raise :class:asyncio.TimeoutError when no matching change occurred within timeout seconds.	None

Returns:

Type	Description
tuple[T, T]	The (value, old_value) pair captured at assignment time.

Source code in src/asyncio_util/_async_value.py

async def wait_transition(
    self,
    value_or_predicate: T | P2 = _any_transition,
    *,
    timeout: float | None = None,
) -> tuple[T, T]:
    """Wait for the next matching change, and return ``(value, old_value)``.

    Unlike :meth:`wait_value`, the current value is never a match:
    this waits for an *edge*, an actual assignment that changes the
    value.

    Args:
        value_or_predicate: A plain value the *new* value must equal,
            or a callable ``predicate(value, old_value) -> bool``.
            Omit it to match any change.
        timeout: If not None, raise :class:`asyncio.TimeoutError`
            when no matching change occurred within *timeout*
            seconds.

    Returns:
        The ``(value, old_value)`` pair captured at assignment time.
    """
    try:
        return await asyncio.wait_for(
            self._wait_predicate(self._edge_results, _ValueWrapper(value_or_predicate)),
            timeout,
        )
    except asyncio.TimeoutError:
        raise asyncio.TimeoutError("Operation timed out") from None

transitions async

transitions(value: T) -> AsyncIterator[tuple[T, T]]

transitions(predicate: P2 = _any_transition) -> AsyncIterator[tuple[T, T]]

transitions(value_or_predicate: T | P2 = _any_transition) -> AsyncIterator[tuple[T, T]]

Iterate over matching changes as (value, old_value) pairs.

Like :meth:wait_transition in a loop: only edges are reported, never the current value.

Warning

Transitions are not queued. A change that happens while the loop body is still processing the previous one is missed. If every edge matters, have the producer push into an asyncio.Queue (or a :class:~asyncio_util.MulticastQueue) instead.

Parameters:

Name	Type	Description	Default
value_or_predicate	T | P2	A plain value the new value must equal, or a callable predicate(value, old_value) -> bool. Omit it to match any change.	_any_transition

Source code in src/asyncio_util/_async_value.py

async def transitions(
    self,
    value_or_predicate: T | P2 = _any_transition,
) -> AsyncIterator[tuple[T, T]]:
    """Iterate over matching changes as ``(value, old_value)`` pairs.

    Like :meth:`wait_transition` in a loop: only edges are reported,
    never the current value.

    Warning:
        Transitions are not queued.  A change that happens while the
        loop body is still processing the previous one is missed.
        If every edge matters, have the producer push into an
        ``asyncio.Queue`` (or a :class:`~asyncio_util.MulticastQueue`)
        instead.

    Args:
        value_or_predicate: A plain value the *new* value must equal,
            or a callable ``predicate(value, old_value) -> bool``.
            Omit it to match any change.
    """
    predicate = _ValueWrapper(value_or_predicate)
    with self._edge_results.open_ref(predicate) as result:
        while True:
            await result.event.park()
            yield result.value

open_transform async

open_transform(function: Callable[[T], T_OUT]) -> AsyncContextManager[AsyncValue[T_OUT]]

Derive a new :class:AsyncValue that tracks function(value).

Within the async with block, the derived value is updated synchronously on every assignment to this value. Updates that produce an equal output are deduplicated by the derived value's own setter. Opening the same function object twice shares one derived value (reference-counted).

Parameters:

Name	Type	Description	Default
function	Callable[[T], T_OUT]	A pure, non-raising function applied to each value. It runs inside the source's setter; if it raises, the exception propagates to the code assigning value (after all other waiters have been notified).	required

Example

av = AsyncValue(3)
async with av.open_transform(lambda x: x * 10) as derived:
    assert derived.value == 30

Source code in src/asyncio_util/_async_value.py

@asynccontextmanager
async def open_transform(
    self, function: Callable[[T], T_OUT]
) -> AsyncContextManager[AsyncValue[T_OUT]]:
    """Derive a new :class:`AsyncValue` that tracks ``function(value)``.

    Within the ``async with`` block, the derived value is updated
    synchronously on every assignment to this value.  Updates that
    produce an equal output are deduplicated by the derived value's
    own setter.  Opening the same function object twice shares one
    derived value (reference-counted).

    Args:
        function: A pure, non-raising function applied to each value.
            It runs inside the source's setter; if it raises, the
            exception propagates to the code assigning ``value``
            (after all other waiters have been notified).

    Example:
        ```python
        av = AsyncValue(3)
        async with av.open_transform(lambda x: x * 10) as derived:
            assert derived.value == 30
        ```
    """
    with self._transforms.open_ref(function) as output:
        if output.value is None:
            output.value = function(self.value)
        yield output

asyncio_util.AsyncBool

Bases: AsyncValue[bool]

Async boolean value with wait/transition capabilities.

Convenience subclass of AsyncValue[bool] with False as default.

Source code in src/asyncio_util/_async_bool.py

class AsyncBool(AsyncValue[bool]):
    """Async boolean value with wait/transition capabilities.

    Convenience subclass of AsyncValue[bool] with False as default.
    """

    def __init__(self, value: bool = False):
        super().__init__(value)

asyncio_util.compose_values

compose_values(*, _transform_: Callable[..., T_OUT] | None = None, **value_map: AsyncValue[Any]) -> _ComposeContext

Compose multiple AsyncValues into a single derived AsyncValue.

Changes to any input AsyncValue automatically update the output.

Usage::

with compose_values(x=val_x, y=val_y) as composite:
    # composite.value is a namedtuple(x=..., y=...)
    await composite.wait_value(lambda v: v.x > 10)

# With transform:
with compose_values(_transform_=lambda v: v.x + v.y,
                    x=val_x, y=val_y) as total:
    await total.wait_value(lambda v: v > 100)

Parameters:

Name	Type	Description	Default
_transform_	Callable[..., T_OUT] | None	Optional function to transform the composite namedtuple into the output value.	None
**value_map	AsyncValue[Any]	Named AsyncValue sources.	{}

Returns:

Type	Description
_ComposeContext	A context manager yielding the composed AsyncValue.

Source code in src/asyncio_util/_compose_values.py

def compose_values(
    *,
    _transform_: Callable[..., T_OUT] | None = None,
    **value_map: AsyncValue[Any],
) -> _ComposeContext:
    """Compose multiple AsyncValues into a single derived AsyncValue.

    Changes to any input AsyncValue automatically update the output.

    Usage::

        with compose_values(x=val_x, y=val_y) as composite:
            # composite.value is a namedtuple(x=..., y=...)
            await composite.wait_value(lambda v: v.x > 10)

        # With transform:
        with compose_values(_transform_=lambda v: v.x + v.y,
                            x=val_x, y=val_y) as total:
            await total.wait_value(lambda v: v > 100)

    Args:
        _transform_: Optional function to transform the composite
            namedtuple into the output value.
        **value_map: Named AsyncValue sources.

    Returns:
        A context manager yielding the composed AsyncValue.
    """
    return _ComposeContext(_transform_, value_map)

asyncio_util.open_held_for async

open_held_for(src: AsyncValue[Any], *, duration: float) -> AsyncIterator[AsyncBool]

Track whether src has been unchanged for duration seconds.

Yields an :class:AsyncBool that becomes True when src has not changed for duration seconds, and resets to False on any change.

Parameters:

Name	Type	Description	Default
src	AsyncValue[Any]	The source value to monitor.	required
duration	float	Seconds the value must remain stable.	required

Source code in src/asyncio_util/_async_value_transforms.py

@asynccontextmanager
async def open_held_for(
    src: AsyncValue[Any], *, duration: float
) -> AsyncIterator[AsyncBool]:
    """Track whether *src* has been unchanged for *duration* seconds.

    Yields an :class:`AsyncBool` that becomes True when *src* has not
    changed for *duration* seconds, and resets to False on any change.

    Args:
        src: The source value to monitor.
        duration: Seconds the value must remain stable.
    """
    output = AsyncBool(False)

    async def _listener() -> None:
        last_value = src.value
        while True:
            try:
                await asyncio.wait_for(
                    src.wait_value(lambda v: v != last_value), timeout=duration
                )
                # Value changed before duration elapsed
                output.value = False
                last_value = src.value
            except asyncio.TimeoutError:
                # Value held for the full duration
                output.value = True
                last_value = src.value
                await src.wait_value(lambda v: v != last_value)
                output.value = False
                last_value = src.value

    task = asyncio.create_task(_listener())
    try:
        yield output
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

asyncio_util.open_hysteresis async

open_hysteresis(src: AsyncValue[bool], *, rising_duration: float = 0, falling_duration: float = 0) -> AsyncIterator[AsyncBool]

Apply hysteresis filtering to a boolean source.

The output transitions from False to True only after src has been True for rising_duration seconds, and from True to False only after src has been False for falling_duration seconds.

Parameters:

Name	Type	Description	Default
src	AsyncValue[bool]	Boolean source value to filter.	required
rising_duration	float	Seconds src must stay True before output rises.	0
falling_duration	float	Seconds src must stay False before output falls.	0

Source code in src/asyncio_util/_async_value_transforms.py

@asynccontextmanager
async def open_hysteresis(
    src: AsyncValue[bool],
    *,
    rising_duration: float = 0,
    falling_duration: float = 0,
) -> AsyncIterator[AsyncBool]:
    """Apply hysteresis filtering to a boolean source.

    The output transitions from False to True only after *src* has
    been True for *rising_duration* seconds, and from True to False
    only after *src* has been False for *falling_duration* seconds.

    Args:
        src: Boolean source value to filter.
        rising_duration: Seconds *src* must stay True before output rises.
        falling_duration: Seconds *src* must stay False before output falls.
    """
    output = AsyncBool(src.value)

    async def _listener() -> None:
        while True:
            anticipated = not output.value
            duration = rising_duration if anticipated else falling_duration

            await src.wait_value(anticipated)

            if duration > 0:
                try:
                    await asyncio.wait_for(
                        src.wait_value(not anticipated), timeout=duration
                    )
                    # Value flipped back before duration elapsed
                    continue
                except asyncio.TimeoutError:
                    pass

            output.value = anticipated

    task = asyncio.create_task(_listener())
    try:
        yield output
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

Tasks and cancellation

asyncio_util.wait_any async

wait_any(*args: Callable[[], Awaitable[Any]]) -> None

Run async callables concurrently, return when the first completes.

All remaining tasks are cancelled after the first one finishes.

Parameters:

Name	Type	Description	Default
*args	Callable[[], Awaitable[Any]]	Zero-argument async callables.	()

Source code in src/asyncio_util/_awaitables.py

async def wait_any(*args: Callable[[], Awaitable[Any]]) -> None:
    """Run async callables concurrently, return when the first completes.

    All remaining tasks are cancelled after the first one finishes.

    Args:
        *args: Zero-argument async callables.
    """
    if not args:
        return

    tasks: list[asyncio.Task[Any]] = []
    try:
        for f in args:
            tasks.append(asyncio.create_task(f()))
        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        await cancel_tasks(pending)
        errors = []
        for task in done:
            try:
                task.result()
            except BaseException as e:
                errors.append(e)
        if errors:
            raise errors[0]
    except BaseException:
        await cancel_tasks(tasks)
        raise

asyncio_util.wait_all async

wait_all(*args: Callable[[], Awaitable[Any]]) -> None

Run async callables concurrently, return when all complete.

If any task raises, all remaining tasks are cancelled and the first exception is re-raised.

Parameters:

Name	Type	Description	Default
*args	Callable[[], Awaitable[Any]]	Zero-argument async callables.	()

Source code in src/asyncio_util/_awaitables.py

async def wait_all(*args: Callable[[], Awaitable[Any]]) -> None:
    """Run async callables concurrently, return when all complete.

    If any task raises, all remaining tasks are cancelled and
    the first exception is re-raised.

    Args:
        *args: Zero-argument async callables.
    """
    if not args:
        return

    tasks: list[asyncio.Task[Any]] = []
    try:
        for f in args:
            tasks.append(asyncio.create_task(f()))
        await asyncio.gather(*tasks)
    except BaseException:
        await cancel_tasks(tasks)
        raise

asyncio_util.wait_any_map async

wait_any_map(*fns: Callable[[], Awaitable[Any]], **fn_map: Callable[[], Awaitable[Any]]) -> Any

Like :func:wait_any, but captures return values from named callables.

Positional callables are fire-and-forget (return values discarded). Keyword callables have their return values captured in a namedtuple.

Returns:

Type	Description
Any	A namedtuple with fields matching the keyword argument names.
Any	Fields for callables that did not finish first are None.

Usage::

result = await wait_any_map(
    timeout=lambda: asyncio.sleep(5),
    value=lambda: some_value.wait_value(predicate),
)
if result.value is not None:
    process(result.value)

Source code in src/asyncio_util/_awaitables.py

async def wait_any_map(
    *fns: Callable[[], Awaitable[Any]],
    **fn_map: Callable[[], Awaitable[Any]],
) -> Any:
    """Like :func:`wait_any`, but captures return values from named callables.

    Positional callables are fire-and-forget (return values discarded).
    Keyword callables have their return values captured in a namedtuple.

    Returns:
        A namedtuple with fields matching the keyword argument names.
        Fields for callables that did not finish first are None.

    Usage::

        result = await wait_any_map(
            timeout=lambda: asyncio.sleep(5),
            value=lambda: some_value.wait_value(predicate),
        )
        if result.value is not None:
            process(result.value)
    """
    if not fns and not fn_map:
        return None

    field_names = list(fn_map.keys())
    if field_names:
        result_type = namedtuple("WaitAnyResults", field_names)  # type: ignore[misc]
        results: dict[str, Any] = {name: None for name in field_names}
    else:
        result_type = None
        results = {}

    all_tasks: list[asyncio.Task[Any]] = []
    task_to_name: dict[asyncio.Task[Any], str | None] = {}

    try:
        for f in fns:
            task = asyncio.create_task(f())
            all_tasks.append(task)
            task_to_name[task] = None

        for name, f in fn_map.items():
            task = asyncio.create_task(f())
            all_tasks.append(task)
            task_to_name[task] = name

        done, pending = await asyncio.wait(
            all_tasks, return_when=asyncio.FIRST_COMPLETED
        )
        await cancel_tasks(pending)

        for task in done:
            name = task_to_name[task]
            if name is not None:
                results[name] = task.result()
            else:
                task.result()

    except BaseException:
        await cancel_tasks(all_tasks)
        raise

    if result_type is not None:
        return result_type(**results)
    return None

asyncio_util.move_on_when

move_on_when(fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> _MoveOnWhen

Cancel the body when fn completes.

Usage::

async with move_on_when(event.wait) as scope:
    await long_running_work()
if scope.cancelled_caught:
    print("interrupted by event")

Parameters:

Name	Type	Description	Default
fn	Callable[..., Awaitable[Any]]	Async callable whose completion triggers cancellation.	required
*args	Any	Positional arguments forwarded to fn.	()
**kwargs	Any	Keyword arguments forwarded to fn.	{}

Returns:

Type	Description
_MoveOnWhen	An async context manager yielding a :class:CancelScope whose
_MoveOnWhen	cancelled_caught attribute indicates whether the body was
_MoveOnWhen	interrupted.

Source code in src/asyncio_util/_move_on_when.py

def move_on_when(
    fn: Callable[..., Awaitable[Any]],
    *args: Any,
    **kwargs: Any,
) -> _MoveOnWhen:
    """Cancel the body when *fn* completes.

    Usage::

        async with move_on_when(event.wait) as scope:
            await long_running_work()
        if scope.cancelled_caught:
            print("interrupted by event")

    Args:
        fn: Async callable whose completion triggers cancellation.
        *args: Positional arguments forwarded to *fn*.
        **kwargs: Keyword arguments forwarded to *fn*.

    Returns:
        An async context manager yielding a :class:`CancelScope` whose
        ``cancelled_caught`` attribute indicates whether the body was
        interrupted.
    """
    return _MoveOnWhen(fn, *args, **kwargs)

asyncio_util.CancelScope

Lightweight cancel scope for move_on_when.

Attributes:

Name	Type	Description
cancelled_caught	bool	True if the body was cancelled by the trigger completing (not by an external cancellation).

Source code in src/asyncio_util/_move_on_when.py

class CancelScope:
    """Lightweight cancel scope for move_on_when.

    Attributes:
        cancelled_caught: True if the body was cancelled by the
            trigger completing (not by an external cancellation).
    """

    def __init__(self):
        self.cancelled_caught: bool = False
        self._should_cancel: bool = False

asyncio_util.run_and_cancelling async

run_and_cancelling(fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> AsyncIterator[None]

Run fn in the background and cancel it when the body exits.

Usage::

async with run_and_cancelling(background_worker, arg1):
    await do_main_work()
# background_worker is cancelled here

If the background task fails while the body is running, the body is not interrupted; the background exception is re-raised when the block exits, unless the body raised its own exception (which takes precedence).

Parameters:

Name	Type	Description	Default
fn	Callable[..., Awaitable[Any]]	Async callable to run in the background.	required
*args	Any	Positional arguments forwarded to fn.	()
**kwargs	Any	Keyword arguments forwarded to fn.	{}

Source code in src/asyncio_util/_move_on_when.py

@asynccontextmanager
async def run_and_cancelling(
    fn: Callable[..., Awaitable[Any]],
    *args: Any,
    **kwargs: Any,
) -> AsyncIterator[None]:
    """Run *fn* in the background and cancel it when the body exits.

    Usage::

        async with run_and_cancelling(background_worker, arg1):
            await do_main_work()
        # background_worker is cancelled here

    If the background task fails while the body is running, the body is
    *not* interrupted; the background exception is re-raised when the
    block exits, unless the body raised its own exception (which takes
    precedence).

    Args:
        fn: Async callable to run in the background.
        *args: Positional arguments forwarded to *fn*.
        **kwargs: Keyword arguments forwarded to *fn*.
    """
    task = asyncio.create_task(fn(*args, **kwargs))
    body_raised = False
    try:
        yield
    except BaseException:
        body_raised = True
        raise
    finally:
        if task.done():
            # Propagate non-cancellation exceptions from the background
            # task — but never mask an exception already propagating
            # from the body.
            try:
                task.result()
            except asyncio.CancelledError:
                pass
            except Exception:
                if not body_raised:
                    raise
        else:
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

asyncio_util.start_and_cancelling async

start_and_cancelling(fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> AsyncIterator[None]

Run fn in the background, wait for it to signal readiness, then cancel it when the body exits.

The callable fn must accept a keyword argument task_status and call task_status.set() when it is ready.

Usage::

async def server(*, task_status):
    listener = await setup_listener()
    task_status.set()  # signal ready
    await serve_forever(listener)

async with start_and_cancelling(server):
    await use_server()

Parameters:

Name	Type	Description	Default
fn	Callable[..., Awaitable[Any]]	Async callable that accepts task_status keyword.	required
*args	Any	Positional arguments forwarded to fn.	()
**kwargs	Any	Keyword arguments forwarded to fn.	{}

Source code in src/asyncio_util/_move_on_when.py

@asynccontextmanager
async def start_and_cancelling(
    fn: Callable[..., Awaitable[Any]],
    *args: Any,
    **kwargs: Any,
) -> AsyncIterator[None]:
    """Run *fn* in the background, wait for it to signal readiness, then
    cancel it when the body exits.

    The callable *fn* must accept a keyword argument ``task_status``
    and call ``task_status.set()`` when it is ready.

    Usage::

        async def server(*, task_status):
            listener = await setup_listener()
            task_status.set()  # signal ready
            await serve_forever(listener)

        async with start_and_cancelling(server):
            await use_server()

    Args:
        fn: Async callable that accepts ``task_status`` keyword.
        *args: Positional arguments forwarded to *fn*.
        **kwargs: Keyword arguments forwarded to *fn*.
    """
    started = asyncio.Event()

    class _TaskStatus:
        def set(self) -> None:
            started.set()

    task_status = _TaskStatus()
    task = asyncio.create_task(fn(*args, task_status=task_status, **kwargs))

    # If the task fails before calling task_status.set(), unblock started.wait()
    task.add_done_callback(lambda _: started.set())

    body_raised = False
    try:
        await started.wait()

        if task.done():
            task.result()  # Propagate exception if task failed before started

        yield
    except BaseException:
        body_raised = True
        raise
    finally:
        if task.done():
            try:
                task.result()
            except asyncio.CancelledError:
                pass
            except Exception:
                if not body_raised:
                    raise
        else:
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

Events, streams and timing

asyncio_util.RepeatedEvent

An event that can be set multiple times, supporting multiple listeners.

Each call to :meth:set increments an internal counter. Listeners can either observe every firing (eventual consistency via :meth:events) or only the latest (drop-while-busy via :meth:unqueued_events).

Source code in src/asyncio_util/_repeated_event.py

class RepeatedEvent:
    """An event that can be set multiple times, supporting multiple listeners.

    Each call to :meth:`set` increments an internal counter.  Listeners
    can either observe every firing (eventual consistency via
    :meth:`events`) or only the latest (drop-while-busy via
    :meth:`unqueued_events`).
    """

    def __init__(self) -> None:
        self._event: AsyncValue[int] = AsyncValue(0)

    def set(self) -> None:
        """Fire the event, waking all current waiters."""
        self._event.value = self._event.value + 1

    async def wait(self) -> None:
        """Wait until the event is fired at least once after this call."""
        token = self._event.value
        await self._event.wait_value(lambda v: v > token)

    async def unqueued_events(self) -> AsyncIterator[None]:
        """Yield on each firing, dropping events that arrive during processing.

        Events that occur while the consumer is processing the previous
        one are silently dropped.
        """
        async for _ in self._event.transitions():
            yield

    async def events(self, *, repeat_last: bool = False) -> AsyncIterator[None]:
        """Yield on each firing with eventual-consistency guarantee.

        The iterator is guaranteed to yield at least once for the most
        recent event, even if intermediate events were missed.

        Args:
            repeat_last: If True, yield immediately for the current
                state before waiting for new events.
        """
        token = -1 if repeat_last else self._event.value
        async for value in self._event.eventual_values(lambda v: v > token):
            token = value
            yield

set

set() -> None

Fire the event, waking all current waiters.

Source code in src/asyncio_util/_repeated_event.py

def set(self) -> None:
    """Fire the event, waking all current waiters."""
    self._event.value = self._event.value + 1

wait async

wait() -> None

Wait until the event is fired at least once after this call.

Source code in src/asyncio_util/_repeated_event.py

async def wait(self) -> None:
    """Wait until the event is fired at least once after this call."""
    token = self._event.value
    await self._event.wait_value(lambda v: v > token)

unqueued_events async

unqueued_events() -> AsyncIterator[None]

Yield on each firing, dropping events that arrive during processing.

Events that occur while the consumer is processing the previous one are silently dropped.

Source code in src/asyncio_util/_repeated_event.py

async def unqueued_events(self) -> AsyncIterator[None]:
    """Yield on each firing, dropping events that arrive during processing.

    Events that occur while the consumer is processing the previous
    one are silently dropped.
    """
    async for _ in self._event.transitions():
        yield

events async

events(*, repeat_last: bool = False) -> AsyncIterator[None]

Yield on each firing with eventual-consistency guarantee.

The iterator is guaranteed to yield at least once for the most recent event, even if intermediate events were missed.

Parameters:

Name	Type	Description	Default
repeat_last	bool	If True, yield immediately for the current state before waiting for new events.	False

Source code in src/asyncio_util/_repeated_event.py

async def events(self, *, repeat_last: bool = False) -> AsyncIterator[None]:
    """Yield on each firing with eventual-consistency guarantee.

    The iterator is guaranteed to yield at least once for the most
    recent event, even if intermediate events were missed.

    Args:
        repeat_last: If True, yield immediately for the current
            state before waiting for new events.
    """
    token = -1 if repeat_last else self._event.value
    async for value in self._event.eventual_values(lambda v: v > token):
        token = value
        yield

asyncio_util.MulticastQueue

Bases: Generic[T]

A broadcast queue that delivers each item to every active listener.

Note

This is a lossy queue: if a listener's internal buffer is full, new messages for that listener are silently dropped. Increase queue_size if you need larger buffers, or consume items faster.

Usage::

mq = MulticastQueue()

async with mq.listen() as q:
    async for item in q:
        process(item)

# In another task:
await mq.broadcast("hello")

Source code in src/asyncio_util/_multicast_queue.py

class MulticastQueue(Generic[T]):
    """A broadcast queue that delivers each item to every active listener.

    Note:
        This is a **lossy** queue: if a listener's internal buffer is full,
        new messages for that listener are silently dropped.  Increase
        *queue_size* if you need larger buffers, or consume items faster.

    Usage::

        mq = MulticastQueue()

        async with mq.listen() as q:
            async for item in q:
                process(item)

        # In another task:
        await mq.broadcast("hello")
    """

    def __init__(self, queue_size: int = 10) -> None:
        self._queue_size = queue_size
        self._listeners: list[asyncio.Queue[Any]] = []

    async def broadcast(self, value: T) -> None:
        """Send *value* to all active listeners."""
        for q in self._listeners:
            try:
                q.put_nowait(value)
            except asyncio.QueueFull:
                pass

    @asynccontextmanager
    async def listen(self) -> AsyncIterator[_Listener[T]]:
        """Register a listener and yield an async iterable of broadcast values."""
        q: asyncio.Queue[Any] = asyncio.Queue(maxsize=self._queue_size)
        self._listeners.append(q)
        try:
            yield _Listener(q)
        finally:
            self._listeners.remove(q)
            # Signal end of iteration; ensure the sentinel is delivered
            # even if the queue is full by dropping an item to make room.
            while True:
                try:
                    q.put_nowait(_CLOSED)
                    break
                except asyncio.QueueFull:
                    try:
                        q.get_nowait()
                    except asyncio.QueueEmpty:
                        break

broadcast async

broadcast(value: T) -> None

Send value to all active listeners.

Source code in src/asyncio_util/_multicast_queue.py

async def broadcast(self, value: T) -> None:
    """Send *value* to all active listeners."""
    for q in self._listeners:
        try:
            q.put_nowait(value)
        except asyncio.QueueFull:
            pass

listen async

listen() -> AsyncIterator[_Listener[T]]

Register a listener and yield an async iterable of broadcast values.

Source code in src/asyncio_util/_multicast_queue.py

@asynccontextmanager
async def listen(self) -> AsyncIterator[_Listener[T]]:
    """Register a listener and yield an async iterable of broadcast values."""
    q: asyncio.Queue[Any] = asyncio.Queue(maxsize=self._queue_size)
    self._listeners.append(q)
    try:
        yield _Listener(q)
    finally:
        self._listeners.remove(q)
        # Signal end of iteration; ensure the sentinel is delivered
        # even if the queue is full by dropping an item to make room.
        while True:
            try:
                q.put_nowait(_CLOSED)
                break
            except asyncio.QueueFull:
                try:
                    q.get_nowait()
                except asyncio.QueueEmpty:
                    break

asyncio_util.periodic async

periodic(period: float) -> AsyncIterator[tuple[float, float | None]]

Yield (elapsed, delta) tuples at regular intervals.

Parameters:

Name	Type	Description	Default
period	float	Interval in seconds between iterations.	required

Yields:

Type	Description
AsyncIterator[tuple[float, float | None]]	A tuple of (elapsed_since_start, delta_since_last).
AsyncIterator[tuple[float, float | None]]	delta is None on the first iteration.

Iterations are scheduled on the absolute start + k * period grid, so per-iteration latency does not accumulate as drift. If the loop body takes longer than period, the next iteration starts (almost) immediately at the next free grid slot; missed slots are skipped, not queued.

Source code in src/asyncio_util/_periodic.py

async def periodic(period: float) -> AsyncIterator[tuple[float, float | None]]:
    """Yield (elapsed, delta) tuples at regular intervals.

    Args:
        period: Interval in seconds between iterations.

    Yields:
        A tuple of (elapsed_since_start, delta_since_last).
        delta is None on the first iteration.

    Iterations are scheduled on the absolute ``start + k * period``
    grid, so per-iteration latency does not accumulate as drift.  If
    the loop body takes longer than *period*, the next iteration starts
    (almost) immediately at the next free grid slot; missed slots are
    skipped, not queued.
    """
    if period <= 0:
        raise ValueError(f"period must be positive, got {period}")
    loop = asyncio.get_running_loop()
    t0 = loop.time()
    t_last: float | None = None
    t_start = t0
    tick = 1

    while True:
        delta = t_start - t_last if t_last is not None else None
        yield (t_start - t0, delta)

        now = loop.time()
        target = t0 + tick * period
        while target <= now:  # overran one or more slots: skip them
            tick += 1
            target = t0 + tick * period
        await asyncio.sleep(target - now)
        tick += 1

        t_last = t_start
        t_start = loop.time()

asyncio_util.azip async

azip(*aiterables: AsyncIterator[T]) -> AsyncIterator[tuple[T, ...]]

Async zip: yield tuples from multiple async iterators in parallel.

Stops when the shortest iterator is exhausted. Advances all iterators concurrently.

Source code in src/asyncio_util/_async_itertools.py

async def azip(
    *aiterables: AsyncIterator[T],
) -> AsyncIterator[tuple[T, ...]]:
    """Async zip: yield tuples from multiple async iterators in parallel.

    Stops when the shortest iterator is exhausted.
    Advances all iterators concurrently.
    """
    if not aiterables:
        return

    iters = [ai.__aiter__() for ai in aiterables]
    sentinel = object()

    while True:
        tasks = [asyncio.create_task(_anext(it, sentinel)) for it in iters]
        try:
            results = await asyncio.gather(*tasks)
        except BaseException:
            await cancel_tasks(tasks)
            raise

        if any(r is sentinel for r in results):
            return
        yield tuple(results)

asyncio_util.azip_longest async

azip_longest(*aiterables: AsyncIterator[T], fillvalue: T | None = None) -> AsyncIterator[tuple[T | None, ...]]

Async zip_longest: yield tuples until all iterators are exhausted.

Exhausted iterators are replaced with fillvalue. Advances all iterators concurrently.

Source code in src/asyncio_util/_async_itertools.py

async def azip_longest(
    *aiterables: AsyncIterator[T],
    fillvalue: T | None = None,
) -> AsyncIterator[tuple[T | None, ...]]:
    """Async zip_longest: yield tuples until all iterators are exhausted.

    Exhausted iterators are replaced with *fillvalue*.
    Advances all iterators concurrently.
    """
    if not aiterables:
        return

    iters = [ai.__aiter__() for ai in aiterables]
    sentinel = object()
    exhausted = [False] * len(iters)

    while True:
        tasks = []
        for i, it in enumerate(iters):
            if exhausted[i]:
                tasks.append(None)
            else:
                tasks.append(asyncio.create_task(_anext(it, sentinel)))

        active_tasks = [t for t in tasks if t is not None]
        try:
            if active_tasks:
                await asyncio.gather(*active_tasks)
        except BaseException:
            await cancel_tasks(active_tasks)
            raise

        results = []
        all_done = True
        for i, t in enumerate(tasks):
            if t is None:
                results.append(fillvalue)
            else:
                val = t.result()
                if val is sentinel:
                    exhausted[i] = True
                    results.append(fillvalue)
                else:
                    all_done = False
                    results.append(val)

        if all_done:
            return
        yield tuple(results)

asyncio_util.iter_move_on_after

Bases: _IterWithTimeout[T]

Wrap an async iterator with a per-item timeout.

If fetching the next item takes longer than timeout seconds, iteration stops silently.

Usage::

async for item in iter_move_on_after(1.0, some_aiter):
    process(item)

Source code in src/asyncio_util/_iterators.py

class iter_move_on_after(_IterWithTimeout[T]):
    """Wrap an async iterator with a per-item timeout.

    If fetching the next item takes longer than *timeout* seconds,
    iteration stops silently.

    Usage::

        async for item in iter_move_on_after(1.0, some_aiter):
            process(item)
    """

    def __init__(self, timeout: float, ait: AsyncIterator[T]) -> None:
        super().__init__(timeout, ait, stop_on_timeout=True)

asyncio_util.iter_fail_after

Bases: _IterWithTimeout[T]

Wrap an async iterator with a per-item timeout.

If fetching the next item takes longer than timeout seconds, raises :class:asyncio.TimeoutError.

Usage::

async for item in iter_fail_after(1.0, some_aiter):
    process(item)

Source code in src/asyncio_util/_iterators.py

class iter_fail_after(_IterWithTimeout[T]):
    """Wrap an async iterator with a per-item timeout.

    If fetching the next item takes longer than *timeout* seconds,
    raises :class:`asyncio.TimeoutError`.

    Usage::

        async for item in iter_fail_after(1.0, some_aiter):
            process(item)
    """

    def __init__(self, timeout: float, ait: AsyncIterator[T]) -> None:
        super().__init__(timeout, ait, stop_on_timeout=False)