Python Cheatsheet 01 — asyncio Essentials

import asyncio

async def main():
    print("hi")

asyncio.run(main())

async def fetch(url):
    await asyncio.sleep(1)
    return url

async def main():
    result = await fetch("a")

async def main():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch("a"))
        b = tg.create_task(fetch("b"))
    print(a.result(), b.result())

results = await asyncio.gather(fetch("a"), fetch("b"))

# With error handling
results = await asyncio.gather(*coros, return_exceptions=True)
for r in results:
    if isinstance(r, Exception): ...

async with asyncio.timeout(5):
    result = await slow()
# raises TimeoutError after 5s

result = await asyncio.wait_for(slow(), timeout=5)

async def worker():
    try:
        await long_op()
    except asyncio.CancelledError:
        await cleanup()
        raise            # always re-raise!

result = await asyncio.to_thread(blocking_io, arg)

from concurrent.futures import ProcessPoolExecutor

pool = ProcessPoolExecutor()

async def compute():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(pool, heavy_function, arg)

sem = asyncio.Semaphore(10)

async def fetch(url):
    async with sem:
        return await http.get(url)

q = asyncio.Queue(maxsize=100)

async def producer():
    for i in range(1000):
        await q.put(i)
    await q.put(None)        # poison pill

async def consumer():
    while True:
        item = await q.get()
        if item is None: break
        await process(item)

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer())
        tg.create_task(consumer())

e = asyncio.Event()

async def waiter():
    await e.wait()
    print("done")

async def signaler():
    await asyncio.sleep(1)
    e.set()

lock = asyncio.Lock()

async def critical():
    async with lock:
        # only one coroutine at a time
        ...

async def bg(): ...

task = asyncio.create_task(bg())
# task runs; result ignored
# WARNING: errors silently swallowed unless you await or add done_callback

def on_done(t):
    if t.exception():
        log.error(t.exception())

task.add_done_callback(on_done)

# If one fails, gather doesn't cancel others by default
results = await asyncio.gather(*tasks, return_exceptions=True)

async def run_with_timeout():
    try:
        async with asyncio.timeout(5):
            return await asyncio.to_thread(blocking, arg)
    except TimeoutError:
        return None