Skip to content

asyncio-util

Utilities for asyncio, ported from trio-util — awaitable values, cancel scopes, repeated events, and more.

asyncio-util brings the ergonomics of trio-util (by GROOVE X) to plain asyncio. Its centerpiece is AsyncValue: a value whose states and transitions any number of tasks can await — no polling loops, no hand-rolled Event juggling, no missed wakeups.

Origin

This project started as the sample code for a PyCon US 2024 talk on porting trio-util to asyncio, and grew into a standalone library.

import asyncio
from asyncio_util import AsyncValue

connection_state = AsyncValue("disconnected")

async def watcher():
    await connection_state.wait_value("connected")
    print("connected!")

async def main():
    task = asyncio.ensure_future(watcher())
    await asyncio.sleep(0)
    connection_state.value = "connected"   # wakes the watcher
    await task

asyncio.run(main())

Why AsyncValue?

Shared state plus asyncio.Event is the usual way to signal between tasks — and it is surprisingly easy to get wrong:

  • An Event carries no value; you must keep the state next to it and keep the two in sync yourself.
  • "Wait until X > 3" requires a loop of check → wait → clear that has to be written carefully to avoid lost wakeups.
  • If the value changes twice before the waiting task runs, the intermediate value is silently lost. A task waiting for 20 misses the moment when the value was 20 and has already moved on.

AsyncValue solves all three: the value and its signaling live in one object, waiting is one await, and matches are evaluated synchronously at assignment time — so a matching change is never missed, even if the value flips back immediately:

av = AsyncValue(10)

# elsewhere: av.value = 20; av.value = 10   (rapid flip)
value = await av.wait_value(20)   # still returns 20

Feature overview

Awaitable values

API What it does
AsyncValue A value whose states and transitions can be awaited
AsyncBool AsyncValue[bool] with a False default
wait_value() Wait until the value matches a value or predicate
wait_transition() Wait for the next matching change (edge)
eventual_values() Iterate over values, always converging to the latest
transitions() Iterate over (new, old) change pairs
open_transform() Derive an AsyncValue tracking f(value)
compose_values() Combine several AsyncValues and await cross-value conditions
open_held_for() An AsyncBool that turns True when a value has been stable
open_hysteresis() Hysteresis filtering for a boolean value

Tasks and cancellation

API What it does
wait_any() Run coroutines concurrently, return on the first to finish
wait_all() Run coroutines concurrently, return when all finish
wait_any_map() wait_any that tells you which finished, with its result
move_on_when() Cancel a block when a trigger completes
run_and_cancelling() Run a background task scoped to a block
start_and_cancelling() Same, but waits for the task to signal readiness

Events, streams and timing

API What it does
RepeatedEvent An event that can fire many times, with multiple listeners
MulticastQueue Broadcast each item to every active listener
periodic() Drift-free periodic iteration
azip() / azip_longest() Async zip over async iterators
iter_move_on_after() / iter_fail_after() Per-item timeouts for async iterators
  • Zero dependencies — only the standard library.
  • Typed — ships a py.typed marker; AsyncValue is generic (AsyncValue[int], AsyncValue[State], …).
  • Python 3.10+, CPython and PyPy.

Installation

Warning

Not yet published to PyPI. Install from GitHub:

pip install git+https://github.com/jrfk/asyncio-util

Where to go next

Acknowledgements

The API is modeled on trio-util by GROOVE X, reimplemented on top of asyncio primitives.