# Async helpers
## Calling async code from sync code
Use these helpers to call async libraries without making your own code
async:
- [`run_sync`](https://fastcore.fast.ai/aio.html#run_sync) runs a
coroutine and returns its result.
- [`iter_sync`](https://fastcore.fast.ai/aio.html#iter_sync) iterates an
async generator.
- [`ctx_sync`](https://fastcore.fast.ai/aio.html#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
``` python
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`](https://fastcore.fast.ai/aio.html#run_sync) on its own
background loop raises `RuntimeError`. Waiting there for a coroutine
would prevent that same loop from running it.
``` python
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`](https://fastcore.fast.ai/aio.html#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`](https://fastcore.fast.ai/aio.html#run_sync):
``` python
async def _outer(): return run_sync(_double(4))
test_eq(await _outer(), 8)
```
------------------------------------------------------------------------
source
### iter_sync
``` python
def iter_sync(
agen
):
```
*Iterate async generator `agen` from sync code*
[`iter_sync`](https://fastcore.fast.ai/aio.html#iter_sync) pulls each
item by running `__anext__` on the shared loop, and closes `agen` when
iteration ends, including when the consumer stops early:
``` python
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
``` python
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`:
``` python
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`](https://fastcore.fast.ai/aio.html#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
``` python
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`:
``` python
import time
from concurrent.futures import ThreadPoolExecutor
```
``` python
@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:
``` python
@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
``` python
async def maybe_await(
o
):
```
*Await `o` if needed, and return it*
``` python
async def _f(): return 42
test_eq(await maybe_await(_f()), 42)
test_eq(await maybe_await('hello'), 'hello')
```
------------------------------------------------------------------------
source
### then
``` python
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`](https://fastcore.fast.ai/aio.html#then) awaits every awaitable
returned by a step. Steps cannot pass awaitables through as data.
``` python
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
``` python
def acache(
f
):
```
*Cache results of async function `f`*
This is like `functools.cache` but for async functions, e.g:
``` python
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)
```
``` python
test_eq([await f(4), await f(4)], [8,8])
test_eq(n, 2)
```
------------------------------------------------------------------------
source
### CachedAwaitable
``` python
def CachedAwaitable(
o
):
```
*Cache the result from an awaitable*
------------------------------------------------------------------------
source
### reawaitable
``` python
def reawaitable(
func:
):
```
*Wraps the result of an asynchronous function into an object which can
be awaited more than once*
`CachedCoro` and
[`reawaitable`](https://fastcore.fast.ai/aio.html#reawaitable) are
partly based on [python issue
tracker](https://bugs.python.org/issue46622) code from Serhiy Storchaka.
They allow an awaitable to be called multiple times.
``` python
@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
``` python
def is_async_callable(
obj
):
```
*Check if `obj` is an async callable, handling `partial` wrappers and
callable instances*
[`is_async_callable`](https://fastcore.fast.ai/aio.html#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](https://github.com/encode/starlette).
``` python
async def f(): pass
assert is_async_callable(f)
assert is_async_callable(partial(f))
assert not is_async_callable(lambda: None)
```
``` python
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
``` python
def to_aiter(
items
):
```
*Async yield each item in `items` with `asyncio.sleep(0)` between*
``` python
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
``` python
def maybe_aiter(
items
):
```
*If `items` already async, return it; otherwise to_aiter*
``` python
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
``` python
async def mapa(
f, items
):
```
*Async `map`; apply `f` (sync or async) to `items` (sync or async iter)
concurrently via `gather`*
``` python
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
``` python
async def noopa(
x:NoneType=None, *args, **kwargs
):
```
*Do nothing (async)*
[`wait_until`](https://fastcore.fast.ai/aio.html#wait_until) repeatedly
calls a sync or async `probe` until it returns a truthy result or the
timeout expires:
------------------------------------------------------------------------
source
### wait_until
``` python
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*
``` python
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:
``` python
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`](https://fastcore.fast.ai/aio.html#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
``` python
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`](https://fastcore.fast.ai/aio.html#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`](https://fastcore.fast.ai/aio.html#athreaded) to avoid
stalling the loop.
Each call resets the `wait` timer:
``` python
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:
``` python
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:
``` python
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`:
``` python
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:
``` python
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:
``` python
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:
``` python
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:
``` python
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:
``` python
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:
``` python
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
``` python
def disable_async_magics(
ip, # An `InteractiveShell` previously passed to `enable_async_magics`
):
```
*Undo
[`enable_async_magics`](https://fastcore.fast.ai/aio.html#enable_async_magics)
on `ip`*
------------------------------------------------------------------------
source
### enable_async_magics
``` python
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.*
``` python
from IPython.core.interactiveshell import InteractiveShell
from fastcore.nbio import run_cell
```
``` python
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`:
``` python
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:
``` python
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`](https://fastcore.fast.ai/aio.html#disable_async_magics)
unwraps the shell:
``` python
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'