# Async helpers


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Calling async code from sync code

Async libraries return coroutines, which normally force every caller up
the stack to become async too.
[`run_sync`](https://fastcore.fast.ai/aio.html#run_sync) (coroutines),
[`iter_sync`](https://fastcore.fast.ai/aio.html#iter_sync) (async
generators), and
[`ctx_sync`](https://fastcore.fast.ai/aio.html#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.

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L42"
target="_blank" style="float:right; font-size:smaller">source</a>

### run_sync

``` python
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`](https://fastcore.fast.ai/aio.html#run_sync) from a
coroutine it is already running would deadlock, so that raises
`RuntimeError` instead.

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

Because the loop lives on its own thread,
[`run_sync`](https://fastcore.fast.ai/aio.html#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:

``` python
async def _outer(): return run_sync(_double(4))
test_eq(asyncio.run(_outer()), 8)
```

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L57"
target="_blank" style="float:right; font-size:smaller">source</a>

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L66"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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) 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.

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L76"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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 genuine coroutine function, so
`asyncio.iscoroutinefunction` (and
[`is_async_callable`](https://fastcore.fast.ai/aio.html#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`:

``` python
import time
from concurrent.futures import ThreadPoolExecutor
```

``` python
@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:

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

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

## Async utilities

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L85"
target="_blank" style="float:right; font-size:smaller">source</a>

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L91"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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*

[`then`](https://fastcore.fast.ai/aio.html#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.

``` 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
```

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L104"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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)
```

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L116"
target="_blank" style="float:right; font-size:smaller">source</a>

### CachedAwaitable

``` python
def CachedAwaitable(
    o
):
```

*Cache the result from an awaitable*

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L124"
target="_blank" style="float:right; font-size:smaller">source</a>

### reawaitable

``` python
def reawaitable(
    func:callable
):
```

*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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L131"
target="_blank" style="float:right; font-size:smaller">source</a>

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L139"
target="_blank" style="float:right; font-size:smaller">source</a>

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L147"
target="_blank" style="float:right; font-size:smaller">source</a>

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L152"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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
```

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L158"
target="_blank" style="float:right; font-size:smaller">source</a>

### noopa

``` python
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`](https://fastcore.fast.ai/aio.html#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_await`](https://fastcore.fast.ai/aio.html#maybe_await)s 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.

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L188"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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`*

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/fastcore/blob/main/fastcore/aio.py#L170"
target="_blank" style="float:right; font-size:smaller">source</a>

### 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

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'

``` python
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
```

``` python
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'