Async and asyncio

import asyncio

async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)          # yields control, does NOT block the loop
    return n * 2

fetch(1)                              # <coroutine object fetch at 0x...>
                                      # RuntimeWarning: never awaited

async def main() -> None:
    result = await fetch(1)           # 2
    print(result)

asyncio.run(main())                   # the one entry point, creates and closes a loop

# gather: ordered results, first exception cancels nothing and propagates
results = await asyncio.gather(fetch(1), fetch(2), fetch(3))     # [2, 4, 6]

# return_exceptions keeps the good results and hands you the failures
results = await asyncio.gather(*coros, return_exceptions=True)
oks = [r for r in results if not isinstance(r, Exception)]

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch(1))
        t2 = tg.create_task(fetch(2))
    # both are done here, results available
    print(t1.result(), t2.result())

# Handle grouped failures with except* (3.11+)
try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(may_fail())
        tg.create_task(may_also_fail())
except* ValueError as eg:
    print("value errors:", eg.exceptions)
except* TimeoutError as eg:
    print("timeouts:", eg.exceptions)

# WRONG, the task may vanish
asyncio.create_task(background_work())

# Right, keep a reference
task = asyncio.create_task(background_work(), name="worker")
...
await task

# Right for fire-and-forget, hold a strong set
_background: set[asyncio.Task] = set()

def spawn(coro):
    t = asyncio.create_task(coro)
    _background.add(t)
    t.add_done_callback(_background.discard)   # and this surfaces exceptions

# Best: use a TaskGroup and let it own the lifetime

# asyncio.timeout: a manager, wraps a whole block (3.11+), the modern choice
try:
    async with asyncio.timeout(5):
        await step_one()
        await step_two()
except TimeoutError:                # builtin TimeoutError since 3.11
    ...

async with asyncio.timeout_at(loop.time() + 10):   # absolute deadline
    ...

# wait_for: one awaitable, cancels it on expiry
result = await asyncio.wait_for(fetch(1), timeout=0.5)

# Results in completion order, not argument order
for future in asyncio.as_completed([fetch(1), fetch(2), fetch(3)]):
    print(await future)

# wait: full control, returns (done, pending)
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
    t.cancel()

# Queue: the standard producer/consumer pipeline, maxsize gives backpressure
q: asyncio.Queue[str | None] = asyncio.Queue(maxsize=100)
await q.put(item)
item = await q.get()
q.task_done()
await q.join()                       # wait until every put item is task_done

# Synchronization, all async context managers
lock = asyncio.Lock()
sem = asyncio.Semaphore(10)          # cap concurrency, the usual rate limiter
event = asyncio.Event()

async with sem:                      # never `with`, that is a TypeError
    await hit_api()

await event.wait()
event.set()

async def worker():
    try:
        await long_running()
    except asyncio.CancelledError:
        await flush_partial_state()      # cleanup
        raise                            # ALWAYS re-raise
    finally:
        await release()

task = asyncio.create_task(worker())
task.cancel()
try:
    await task                           # wait for it to actually finish
except asyncio.CancelledError:
    pass

# Shield work that must not be cancelled (use sparingly)
await asyncio.shield(commit_transaction())

import asyncio, time

# WRONG
async def bad():
    time.sleep(1)                  # the entire program is frozen for 1s
    data = requests.get(url).text  # same problem

# Right: push blocking I/O to a thread (3.9+)
async def good():
    data = await asyncio.to_thread(requests.get, url)
    text = await asyncio.to_thread(path.read_text)

# Right: an explicit executor, and the only way to reach processes
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=8) as pool:
    result = await loop.run_in_executor(pool, blocking_io, arg)

with ProcessPoolExecutor() as pool:             # CPU-bound work, dodges the GIL
    result = await loop.run_in_executor(pool, heavy_math, arg)

import asyncio, httpx

async def fetch_one(client: httpx.AsyncClient, url: str,
                    sem: asyncio.Semaphore) -> tuple[str, int]:
    async with sem:                                   # cap in-flight requests
        r = await client.get(url, timeout=10.0)
        return url, r.status_code

async def fetch_all(urls: list[str], limit: int = 20) -> list[tuple[str, int]]:
    sem = asyncio.Semaphore(limit)
    async with httpx.AsyncClient(http2=True) as client:   # ONE client, reused
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(client, u, sem)) for u in urls]
    return [t.result() for t in tasks]

results = asyncio.run(fetch_all(urls))

# aiohttp is the other common choice
import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.get(url) as resp:
        payload = await resp.json()

# async generator: async def + yield
async def paginate(client, url):
    while url:
        page = await client.get(url)
        for item in page["items"]:
            yield item
        url = page.get("next")

async for item in paginate(client, "/api/items"):
    process(item)

# Async comprehension
rows = [r async for r in paginate(client, url) if r["ok"]]

# The protocol behind it
class Ticker:
    def __aiter__(self):
        return self
    async def __anext__(self):
        await asyncio.sleep(1)
        if self.done:
            raise StopAsyncIteration
        return self.tick()

# Close an async generator deterministically
from contextlib import aclosing
async with aclosing(paginate(client, url)) as gen:
    async for item in gen:
        ...

asyncio.run(main(), debug=True)      # warns on slow callbacks and never-awaited coros

PYTHONASYNCIODEBUG=1 python app.py
python -X dev app.py                 # dev mode, same warnings plus more

# What is running right now
for t in asyncio.all_tasks():
    print(t.get_name(), t.get_coro())
asyncio.current_task().get_name()

# Full stack of every task, the fastest way to find a hang
for t in asyncio.all_tasks():
    t.print_stack()

# python -m asyncio (3.8+) is an async REPL where top-level `await` works