Async helpers

Bridging async and sync code: run_sync, iter_sync, ctx_sync, athreaded, maybe_await, and then

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.

---

source

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). Here we simulate that by calling it from inside a coroutine:

async def _outer(): return run_sync(_double(4))
test_eq(asyncio.run(_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 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.

---

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 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.1)
    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.18, 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

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

---

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

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, and 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 – shells enabled twice (e.g. by two extensions sharing this machinery) get exactly one wrap – and lives on the shell rather than user_ns, so %reset can’t break it.

---

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

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(ip.run_cell('%amag hi').result, 'async:hi')
test_eq(ip.run_cell('%smag hi').result, 'sync:hi')      # sync magics unaffected
test_eq(ip.run_cell('%%acell top\nbody').result, 'top:body')
ip.run_cell('x = %amag val')                            # assignment form works too
test_eq(ip.user_ns['x'], 'async:val')

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

enable_async_magics(ip)  # second enable is a no-op: exactly one wrap
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')  # function bodies stay sync

ip2 = InteractiveShell()
ip2.register_magic_function(_amag, 'line', 'amag')
enable_async_magics(ip2, fmt=str.upper)
test_eq(ip2.run_cell('%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'