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

Use these helpers to call async libraries without making your own code async:

• run_sync runs a coroutine and returns its result.
• iter_sync iterates an async generator.
• ctx_sync uses an async context manager in a with block.

They share an event loop on a background daemon thread, started on first use.

---

source

run_sync

def run_sync(
    coro
):

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

Exceptions from the coroutine propagate to the caller. Ctrl-C cancels the submitted coroutine. Calling run_sync on its own background loop raises RuntimeError. Waiting there for a coroutine would prevent that same loop from running it.

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())

You can call run_sync from a thread with a running event loop, such as a Jupyter cell. The coroutine runs on the shared background loop. Calling asyncio.run in that situation would raise RuntimeError. This async function can call run_sync:

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

---

source

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])

---

source

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 turns a blocking function into an async function that runs its body in a worker thread. This lets async code use blocking libraries without stalling the event loop.

---

source

athreaded

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

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

The wrapper is a coroutine function and works as a method. It uses the awaiting loop’s default executor and copies contextvars into the worker thread. Exceptions propagate to the awaiting task with their tracebacks.

Calls can run concurrently 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

---

source

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')

---

source

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

await maybe_await(o) awaits o if it is awaitable and returns it unchanged otherwise.

then(x, *fs) applies each function to the preceding result. It returns a plain value when all steps are synchronous, or an awaitable if x or any step is awaitable.

This is useful for clients with both sync and async modes. A method can use return then(self.gists.get(gid), ~Self.files.values(), first) in either mode, without duplicating its body.

then awaits every awaitable returned by a step. Steps cannot pass awaitables through as data.

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

---

source

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)

---

source

CachedAwaitable

def CachedAwaitable(
    o
):

Cache the result from an awaitable

---

source

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

---

source

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())

---

source

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([])], [])

---

source

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])

---

source

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

---

source

noopa

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

Do nothing (async)

wait_until repeatedly calls a sync or async probe until it returns a truthy result or the timeout expires:

---

source

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 delays a function call until wait seconds after calls stop arriving. For example, use it to autosave after editing pauses. Set max_wait to limit how long repeated calls can postpone it.

Calls take no arguments. Keep pending edits or queued items in your own state, and consume them in f.

---

source

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 returns immediately and schedules f on its event loop. The instance keeps the loop it was constructed on, or the first loop it is called from.

f can be sync or async. Wrap slow blocking work with athreaded to avoid stalling the loop.

Each call resets the wait timer:

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])

Calls every 0.06 seconds keep resetting the 0.1-second timer. f runs once after the calls stop:

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])

max_wait sets a deadline relative to the start of the burst. Repeated calls reset the wait timer without extending this deadline. Setting max_wait=wait limits f to one run per wait seconds. The following calls never pause for wait seconds. The deadline lets f run before the stream stops:

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)

Calls carry no data. The caller decides whether f uses the latest state or every queued item. Update a saved value for the first case, or accumulate items for the second. This example keeps all queued items and drains them 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])

By default, an exception in f ends the worker task and discards pending calls. Asyncio reports the unhandled exception.

With on_error, the handler receives the exception and the worker continues. Pending calls still run, and f can schedule a retry by calling the debounce again:

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 pending work immediately, for example at shutdown. It does nothing when no work is pending. Exceptions propagate to its caller, bypassing on_error.

Call flush and cancel only from the owner loop:

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])

After the instance has an owner loop, calls from other threads are forwarded to it, even if those threads have their own 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])

Calls to the debounce while an async f is running return immediately and request one further run. That run waits for the current run to finish; the two never overlap:

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

enable_async_magics(ip) lets an IPython shell await async line and cell magics. Sync magics continue to work normally. Only unindented magic calls are awaited; calls inside functions are unchanged.

Enabling it twice has no additional effect, and %reset does not disable it.

The transformation runs after IPython converts magic syntax to function calls. This requires wrapping transform_cell, since line_transforms run before line magic conversion.

---

source

disable_async_magics

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

Undo enable_async_magics on ip

---

source

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'

Synchronous magics return their usual values. An assignment such as x = %amag val stores the awaited result in x:

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'