Async helpers

Bridging async and sync code: run_sync, iter_sync, ctx_sync, athreaded, maybe_await, and then, plus Debounce for coalescing bursts of calls

Calling async code from sync code

Async libraries return coroutines, which normally force every caller up the stack to become async too. run_sync (coroutines), iter_sync (async generators), and ctx_sync (async context managers) let ordinary sync code drive them instead, sharing a single event loop that runs on a background daemon thread, created on first use; maybe_await(o) awaits o only if it’s awaitable, else returns it as-is.

run_sync

def run_sync(
    coro
):

Run coroutine coro to completion from sync code and return its result

Exceptions raised inside the coroutine propagate to the caller, and Ctrl-C cancels it rather than leaving it running on the loop thread. Calling run_sync from a coroutine it is already running would deadlock, so that raises RuntimeError instead.

async def _double(x):
    await asyncio.sleep(0.01)
    return x*2

test_eq(run_sync(_double(3)), 6)

async def _boom(): raise ValueError('boom')
with expect_fail(Exception, 'boom'): run_sync(_boom())

async def _nested(): return run_sync(_double(1))
with expect_fail(Exception, 'its own event loop'): run_sync(_nested())

Because the loop lives on its own thread, run_sync also works where the calling thread already has a running loop, such as a Jupyter cell (where asyncio.run raises RuntimeError). A test cell like this one runs on exactly such a loop:

async def _outer(): return run_sync(_double(4))
test_eq(await _outer(), 8)

iter_sync

def iter_sync(
    agen
):

Iterate async generator agen from sync code

iter_sync pulls each item by running __anext__ on the shared loop, and closes agen when iteration ends, including when the consumer stops early:

done = []
async def _agen():
    try:
        i = 0
        while True:
            yield i
            i += 1
    finally: done.append(True)

it = iter_sync(_agen())
test_eq([next(it) for _ in range(3)], [0,1,2])
it.close()
test_eq(done, [True])

ctx_sync

def ctx_sync(
    acm
):

Use async context manager acm in a plain with block

__aenter__ and __aexit__ run on the shared loop, and exception suppression behaves as it would under async with:

events = []
@contextlib.asynccontextmanager
async def _actx():
    events.append('enter')
    try: yield 'ready'
    finally: events.append('exit')

with ctx_sync(_actx()) as v: test_eq(v, 'ready')
test_eq(events, ['enter','exit'])

Calling sync code from async code

athreaded is the same bridge crossed in the other direction: it wraps a blocking function as an honest async def, running the original body in a worker thread so the event loop stays free. Use it to give async color to APIs that only exist in blocking form.

athreaded

def athreaded(
    f, *, executor:NoneType=None
):

Run f in a worker thread, awaitably; use as @athreaded or @athreaded(executor=...)

The wrapper is a genuine coroutine function, so asyncio.iscoroutinefunction (and is_async_callable, below) reports it as async, and it binds as a method like any plain function. The body runs on the awaiting loop’s default executor, contextvars propagate into the thread (as asyncio.to_thread does), and an exception raises in the awaiting task with its real traceback. Because each call occupies a thread rather than the loop, calls overlap under asyncio.gather:

import time
from concurrent.futures import ThreadPoolExecutor

@athreaded
def _slow(x):
    time.sleep(0.25)
    return x*2

assert asyncio.iscoroutinefunction(_slow)
start = time.time()
test_eq(await asyncio.gather(_slow(1), _slow(2)), [2,4])
elapsed = time.time()-start
assert elapsed < 0.45, elapsed
elapsed

Pass executor= for a dedicated pool, so long-blocking calls don’t queue behind whatever else the loop’s shared default pool is running:

@athreaded(executor=ThreadPoolExecutor(thread_name_prefix='ded'))
def _where(): return threading.current_thread().name

assert (await _where()).startswith('ded')
await _where()

Async utilities

maybe_await

async def maybe_await(
    o
):

Await o if needed, and return it

async def _f(): return 42
test_eq(await maybe_await(_f()), 42)
test_eq(await maybe_await('hello'), 'hello')

then

def then(
    x, *fs
):

Pipe x through each of fs, awaiting values as needed; result is awaitable only if x or a step result is

then lets one function body serve sync and async callers alike. It applies each of fs in turn, awaiting anything awaitable along the way (the starting value or any step’s result), and the caller gets back a plain value if the whole chain was sync, or an awaitable otherwise. This is what a convenience method on a client offering both sync and async modes needs: written as return then(self.gists.get(gid), ~Self.files.values(), first), a sync client returns the file directly while an async client returns something to await, with no duplicated method. Because step results are awaited as part of the chain, steps may themselves be async calls; the flip side is that a step cannot return an awaitable as its value.

async def _f(): return dict(a=1,b=2)
test_eq(then(dict(a=1,b=2), ~Self.values(), first), 1)  # all sync: plain result
test_eq(await then(_f(), ~Self.values(), first), 1)     # awaitable start
async def _double(x): return x*2
test_eq(await then(3, _double, mul(10)), 60)            # awaitable appears mid-chain
test_eq(then(3), 3)
test_eq(await then(_f()), dict(a=1,b=2))                # no steps: pure passthrough

acache

def acache(
    f
):

Cache results of async function f

This is like functools.cache but for async functions, e.g:

n = 0

@acache
async def f(x):
    global n
    n += 1
    return x*2

test_eq([await f(3), await f(3)], [6,6])
test_eq(n, 1)

test_eq([await f(4), await f(4)], [8,8])
test_eq(n, 2)

CachedAwaitable

def CachedAwaitable(
    o
):

Cache the result from an awaitable

reawaitable

def reawaitable(
    func:<built-in function callable>
):

Wraps the result of an asynchronous function into an object which can be awaited more than once

CachedCoro and reawaitable are partly based on python issue tracker code from Serhiy Storchaka. They allow an awaitable to be called multiple times.

@reawaitable
async def fetch_data():
    await asyncio.sleep(0.1)
    return "data"

r = fetch_data()
print(await r)  # "data"
print(await r)  # "data" (no delay)

data
data

is_async_callable

def is_async_callable(
    obj
):

Check if obj is an async callable, handling partial wrappers and callable instances

is_async_callable detects whether an object can be called asynchronously. It goes beyond asyncio.iscoroutinefunction by also handling functools.partial-wrapped async functions (unwrapping through any number of layers) and callable objects whose __call__ method is a coroutine. The implementation is thanks to Starlette.

async def f(): pass
assert is_async_callable(f)
assert is_async_callable(partial(f))
assert not is_async_callable(lambda: None)

class AsyncObj:
    async def __call__(self): pass

class SyncObj:
    def __call__(self): pass

assert is_async_callable(AsyncObj())
assert not is_async_callable(SyncObj())

to_aiter

def to_aiter(
    items
):

Async yield each item in items with asyncio.sleep(0) between

test_eq([o async for o in to_aiter([10,20,30])], [10,20,30])
test_eq([o async for o in to_aiter([])], [])

maybe_aiter

def maybe_aiter(
    items
):

If items already async, return it; otherwise to_aiter

async def _agen():
    for i in [1,2,3]: yield i

ag = _agen()
test_eq([o async for o in maybe_aiter(ag)], [1,2,3])
test_eq([o async for o in maybe_aiter([1,2,3])], [1,2,3])

mapa

async def mapa(
    f, items
):

Async map; apply f (sync or async) to items (sync or async iter) concurrently via gather

async def _double(x): return x*2

test_eq(await mapa(mul(2), [1,2,3]), [2,4,6])       # sync f, sync items
test_eq(await mapa(_double, [1,2,3]), [2,4,6])              # async f, sync items
test_eq(await mapa(mul(2), _agen()), [2,4,6])        # sync f, async items
test_eq(await mapa(_double, _agen()), [2,4,6])              # async f, async items

noopa

async def noopa(
    x:NoneType=None, *args, **kwargs
):

Do nothing (async)

Every “wait for X” helper is the same loop: call a probe, sleep, give up at a deadline. wait_until is that loop, taking a sync or async probe and returning its first truthy result:

wait_until

async def wait_until(
    probe, # Sync or async callable; falsy result means keep waiting
    what:str=None, # Named in the `TimeoutError`; `probe`'s name if None
    timeout:float=10, # Seconds to wait before raising
    sleep:float=0.2, # Seconds between probes
):

Call probe until truthy, returning its first truthy result

n = 0
def _probe():
    global n
    n += 1
    return n if n>2 else 0
test_eq(await wait_until(_probe, sleep=0.01), 3)
async def _ready(): return n>2
assert await wait_until(_ready, sleep=0.01)

On timeout it raises TimeoutError naming what, or the probe itself when what is omitted:

with expect_fail(TimeoutError, contains='login form'): await wait_until(lambda: False, 'login form', timeout=0.05, sleep=0.01)
def _never(): return False
with expect_fail(TimeoutError, contains='_never'): await wait_until(_never, timeout=0.05, sleep=0.01)

Debounce

Debounce coalesces a burst of calls into one invocation of f: wait secs after calls stop arriving, or max_wait secs after the burst began. A typical use is auto-saving shortly after edits stop, rather than on every edit. Calls carry no arguments: Debounce is a trigger, and whatever state the fire should consume (pending edits, queued items) lives with the caller, which accumulates it before each call and drains it inside f.

Debounce

def Debounce(
    f, wait, max_wait:NoneType=None, on_error:NoneType=None
):

Coalesce a burst of calls into one run of f, wait secs after calls stop (or max_wait secs after they start)

Calling a Debounce is fire-and-forget: the call returns at once, and f runs later in a worker task on the event loop. The instance binds to the loop it was constructed on, or else the first loop it is called from, and keeps that loop for life. f can be sync or async, but a blocking f will stall the loop, so wrap slow sync work with athreaded or similar. Each call resets the wait timer, so a steady stream of calls fires nothing until they pause:

fires = []
d = Debounce(lambda: fires.append(1), 0.1)
for i in range(5): d()
test_eq(fires, [])
await asyncio.sleep(0.25)
test_eq(fires, [1])

The reset also holds for spaced calls: gaps shorter than wait keep postponing the fire, so only the pause at the end lets it through:

fires.clear()
d = Debounce(lambda: fires.append(1), 0.1)
for i in range(4):
    d()
    await asyncio.sleep(0.06)
test_eq(fires, [])
await asyncio.sleep(0.15)
test_eq(fires, [1])

With max_wait set, a burst cannot postpone the fire forever: it comes no later than max_wait after the burst began. max_wait=wait gives throttle timing, at most one fire per wait secs. The stream below never pauses long enough for a plain debounce to fire, but the cap forces fires mid-stream, and the count shows how few runs the burst cost:

fires.clear()
d = Debounce(lambda: fires.append(1), 0.1, max_wait=0.2)
for i in range(8):
    d()
    await asyncio.sleep(0.06)
assert fires
await asyncio.sleep(0.25)
len(fires)

Because calls carry no payload, “latest args win” and “keep every item” are both caller idioms, not modes. Accumulate into whatever structure suits, and drain it in f:

saved,pending = [],[]
def _drain(): saved.extend(pending); pending.clear()
d = Debounce(_drain, 0.1)
for i in range(5): pending.append(i); d()
await asyncio.sleep(0.15)
test_eq(saved, [0,1,2,3,4])

If f raises, by default the exception ends that burst’s worker task (asyncio logs it as an unretrieved task exception), and any call that arrived during the failing fire is lost with it. Pass on_error to handle the exception instead: the worker survives, so pending calls still fire, and f may retrigger itself to retry:

fires,errs = [],[]
def _boom():
    if not fires: fires.append('bad'); raise ValueError('boom')
    fires.append('good')

d = Debounce(_boom, 0.05, on_error=errs.append)
d()
await asyncio.sleep(0.1)
d()
await asyncio.sleep(0.1)
test_eq(fires, ['bad','good'])
test_eq(len(errs), 1)

flush runs any pending fire immediately, which is handy at shutdown, and propagates exceptions to its caller rather than to on_error. With nothing pending it is a no-op. Use it, like cancel, from the owner loop only:

fires.clear()
d = Debounce(lambda: fires.append(1), 5)
d()
await d.flush()
test_eq(fires, [1])
await d.flush()
test_eq(fires, [1])

cancel discards the pending fire:

d()
d.cancel()
await asyncio.sleep(0.05)
test_eq(fires, [1])

Calls are safe from threads with no loop, and from threads running a different loop: both are forwarded to the owner loop with call_soon_threadsafe, and the instance never migrates between loops:

fires.clear()
d = Debounce(lambda: fires.append(1), 0.05)
threading.Thread(target=d).start()
await asyncio.sleep(0.15)
test_eq(fires, [1])

An async f never overlaps itself: calls that arrive while it is still running wait for it to finish, then trigger one further fire:

fires = []
started,release = asyncio.Event(),asyncio.Event()
async def _slow():
    fires.append(1)
    if len(fires)==1:
        started.set()
        await release.wait()

d = Debounce(_slow, 0.05)
d()
await started.wait()
d()
d()
await asyncio.sleep(0.1)
test_eq(fires, [1])

Releasing the stalled fire lets the one coalesced follow-up run:

release.set()
await asyncio.sleep(0.1)
test_eq(fires, [1,1])

Async magics

IPython magics are ordinary functions, so a magic defined async def hands its caller an unawaited coroutine. enable_async_magics fixes that per shell: it wraps the shell’s input transformation so a top-level magic call in the transformed source is awaited through a per-shell _amagic hook, which maybe_awaits the result, so sync magics are unaffected. The rewrite must see get_ipython().run_line_magic(...) form. Line magics only reach that form in the token transform phase, after every line_transforms entry has run, so this wraps transform_cell itself rather than joining line_transforms. Only unindented magic lines are wrapped: a magic inside a function body stays sync, since await would be a syntax error there. The _amagic attribute doubles as the idempotence sentinel, so shells enabled twice (say, by two extensions sharing this machinery) get exactly one wrap. It lives on the shell rather than user_ns, so %reset can’t break it.

disable_async_magics

def disable_async_magics(
    ip, # An `InteractiveShell` previously passed to `enable_async_magics`
):

Undo enable_async_magics on ip

enable_async_magics

def enable_async_magics(
    ip, # An `InteractiveShell`
    fmt:NoneType=None, # Optional post-await result transform, e.g. FT->HTML conversion
):

Let line and cell magics on ip be async: coroutine results are awaited. Idempotent per shell.

from IPython.core.interactiveshell import InteractiveShell
from fastcore.nbio import run_cell

ip = InteractiveShell()
async def _amag(line): return f'async:{line}'
async def _acell(line, cell): return f'{line}:{cell.strip()}'
ip.register_magic_function(_amag, 'line', 'amag')
ip.register_magic_function(lambda line: f'sync:{line}', 'line', 'smag')
ip.register_magic_function(_acell, 'cell', 'acell')
enable_async_magics(ip)
test_eq((await run_cell(ip, '%amag hi')).result, 'async:hi')
test_eq((await run_cell(ip, '%%acell top\nbody')).result, 'top:body')

Out[0]: 'async:hi'
Out[0]: 'sync:hi'
Out[0]: 'top:body'

The wrap only awaits what needs awaiting, so a sync magic runs exactly as before, and an assignment target on the magic line survives the rewrite:

test_eq((await run_cell(ip, '%smag hi')).result, 'sync:hi')
await run_cell(ip, 'x = %amag val')
test_eq(ip.user_ns['x'], 'async:val')

Enabling twice wraps exactly once, and function bodies stay sync:

enable_async_magics(ip)
test_eq(ip.input_transformer_manager.transform_cell('%smag hi').count('_amagic'), 1)
assert 'await' not in ip.input_transformer_manager.transform_cell('def f():\n    %smag hi')

fmt post-processes each awaited magic result, and disable_async_magics unwraps the shell:

ip2 = InteractiveShell()
ip2.register_magic_function(_amag, 'line', 'amag')
enable_async_magics(ip2, fmt=str.upper)
test_eq((await run_cell(ip2, '%amag hi')).result, 'ASYNC:HI')
disable_async_magics(ip2)
assert 'await' not in ip2.input_transformer_manager.transform_cell('%amag hi')

Out[0]: 'ASYNC:HI'