python-async

Pair with python-common-practice for shared verification guidance.

Persona: You are a Python async engineer. You treat the event loop as a single thread you must never block, and you make concurrency explicit.

Asynchronous Python

async/await gives concurrency for I/O-bound work (HTTP, DB, files) on one thread. It does nothing for CPU-bound work — a tight loop inside async def freezes the whole loop. Reach for async when you wait on the network, not when you crunch numbers.

The cardinal rule: never block the loop

A single synchronous call inside a coroutine stalls every other task. Diagnose these with ruff's ASYNC rules (→ See python-lint).

# ✗ Bad — blocks the event loop; all concurrent tasks freeze
async def fetch(url):
    return requests.get(url).json()          # sync I/O
    time.sleep(5)                            # sync sleep

# ✓ Good — yields control while waiting
async def fetch(client: httpx.AsyncClient, url):
    resp = await client.get(url)
    return resp.json()
    # await asyncio.sleep(5)  for delays

For unavoidable blocking work (a sync library, CPU work), offload it: await asyncio.to_thread(blocking_fn, arg). This keeps the loop responsive.

Run things concurrently — explicitly

awaiting calls one after another is sequential. To overlap, create tasks. Prefer asyncio.TaskGroup (3.11+): it cancels siblings on failure and propagates errors, so you never leak orphaned tasks.

async def fetch_all(client, ids):
    async with asyncio.TaskGroup() as tg:        # structured concurrency
        tasks = [tg.create_task(fetch(client, i)) for i in ids]
    return [t.result() for t in tasks]           # all done, errors already raised

asyncio.gather(*coros) still works but, unless return_exceptions=True, a failure leaves the other tasks running unawaited. Prefer TaskGroup for new code.

Bound concurrency — don't DoS the upstream

Unbounded gather over thousands of items opens thousands of sockets and gets you rate-limited. Cap it with a Semaphore.

async def fetch_bounded(client, ids, limit=10):
    sem = asyncio.Semaphore(limit)
    async def one(i):
        async with sem:                  # at most `limit` in flight
            return await fetch(client, i)
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(one(i)) for i in ids]
    return [t.result() for t in tasks]

This mirrors sortarr's pipeline_concurrency setting — a configurable cap, not unbounded fan-out.

Timeouts and cancellation

Every network call needs a deadline, or one hung upstream hangs your service. Use asyncio.timeout (3.11+) as a context manager; it cancels the body on expiry.

async with asyncio.timeout(30):
    data = await fetch(client, url)

Cancellation arrives as asyncio.CancelledError. Never swallow it — let it propagate so structured concurrency can unwind. If you must clean up, re-raise:

try:
    await long_op()
except asyncio.CancelledError:
    await cleanup()
    raise                         # ✓ propagate; swallowing breaks cancellation

httpx — reuse one client

Create one AsyncClient per app/lifespan and share it; a fresh client per request throws away connection pooling and TLS reuse. Always close it (or use it as a context manager).

async with httpx.AsyncClient(timeout=30) as client:
    ...        # reuse across many requests

Background scheduling (APScheduler)

For periodic jobs, use AsyncIOScheduler so jobs share the app's loop rather than spawning threads. Job functions are coroutines; keep them idempotent (a missed/overlapping run shouldn't corrupt state) and let misfire_grace_time/max_instances=1 guard against pile-ups.

Common mistakes

Diagnose: 1- ruff check with the ASYNC rule set — finds blocking calls in coroutines 2- run with PYTHONASYNCIODEBUG=1 — surfaces un-awaited coroutines and slow callbacks 3- py-spy dump --pid <pid> — see where a hung event loop is stuck.

References

• asyncio — Asynchronous I/O
• Coroutines and Tasks
• httpx — Async usage
• APScheduler documentation