atelier/languages/py-async.md

Python Async — Derived Application

Applies Atelier's domain principles to Python async specifically. Derives from domains/ docs; introduces no new P-rules (D-063). See languages/python.md for the language first-principles stub.

asyncio and anyio (Concurrency P7 Cancellation Support, C2 Clarity)

• asyncio for I/O-bound work; threads only for blocking libraries: async def + await for network/disk; run_in_executor to wrap a blocking call. Mixing threads for I/O is the wrong default.
• anyio for runtime portability: anyio abstracts asyncio/trio; a library written against anyio runs on either. Use it for libraries; for applications, asyncio directly is fine.
• One event loop, one thread: asyncio.run(main()) creates and runs the loop. Do not call asyncio.run inside an existing loop (raises RuntimeError); do not share a loop across threads.
• Applies concurrency/P7: every async def accepts cancellation as a first-class signal; CancelledError propagates unless explicitly suppressed (and suppressing it is almost always a bug).

import asyncio
import anyio

async def fetch_user(id: str) -> User:
    return await api.get(f'/users/{id}')

# asyncio application
async def main():
    user = await fetch_user('abc')

asyncio.run(main())

# anyio library — portable across asyncio/trio
async def fetch_all(ids: list[str]) -> list[User]:
    return await anyio.gather(*(fetch_user(i) for i in ids))

Structured Concurrency (Concurrency P1 Immutability by Default, C6 Composability)

• asyncio.TaskGroup (3.11+) for structured concurrency: tasks created in a TaskGroup are awaited or cancelled together on exit. No orphan tasks outlive the block.
• No asyncio.gather(..., return_exceptions=False) for fallible tasks: gather returns partial results on first exception; TaskGroup cancels siblings and propagates the error atomically. Use TaskGroup for new code.
• Applies concurrency/P1 (immutability): tasks share only immutable inputs; results are collected, not mutated in place. A task that writes to a shared list is a race waiting to happen.
• anyio.create_task_group() mirrors TaskGroup cross-runtime: same structured-concurrency guarantee, portable.

import asyncio

async def fetch_all(ids: list[str]) -> list[User]:
    results: list[User] = []
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user(i)) for i in ids]
    # all tasks done (or cancelled) by here
    return [t.result() for t in tasks]

Cancellation and Timeout (Concurrency P7 Cancellation Support, Concurrency P8 Timeout Discipline)

• asyncio.wait_for(coro, timeout) for a deadline: every external await races against a timeout. A bare await is an unbounded wait (Concurrency P8).
• asyncio.timeout() (3.11+) as a context manager: async with asyncio.timeout(5): await op — cleaner than wait_for for multi-await blocks.
• CancelledError propagates; do not catch broadly: except Exception swallows CancelledError in 3.7 (it was BaseException); in 3.8+ it's BaseException and except Exception skips it. Catch specifically, never bare except:.
• Applies concurrency/P7: cancellation is cooperative — a long synchronous block inside async def ignores cancellation. Yield with await asyncio.sleep(0) periodically in CPU-bound loops.

import asyncio

async def fetch_with_timeout(id: str, timeout: float = 5.0) -> User:
    async with asyncio.timeout(timeout):
        return await fetch_user(id)

async def shutdown(token: asyncio.Event) -> None:
    # cooperative cancel — long-running loop checks the token
    while not token.is_set():
        await do_chunk()
        await asyncio.sleep(0)   # yield so cancel can land

Bounded Concurrency and Queues (Concurrency P9 Bounded Queues)

• asyncio.Semaphore(N) to bound in-flight tasks: a Semaphore(8) wrapping gather caps concurrency. Unbounded gather on a 10k-item list exhausts file descriptors (Concurrency P9 — bounded queues).
• asyncio.Queue(maxsize=N) for producer/consumer: a bounded queue applies backpressure to the producer. An unbounded queue lets the producer run ahead and OOM.
• Applies messaging/queues: an asyncio.Queue is an in-process broker — the same bounded-queue / backpressure semantics apply; the broker is just local.

import asyncio

async def map_bounded(items: list[str], limit: int = 8) -> list[User]:
    sem = asyncio.Semaphore(limit)
    async def guarded(i: str) -> User:
        async with sem:
            return await fetch_user(i)
    return await asyncio.gather(*(guarded(i) for i in items))

Error Handling in Async (Errors P5 Recoverable When Possible, Errors P1 Errors are Data)

• Retry with backoff for transient failures: network blips are recoverable (Errors P5). Exponential backoff with jitter, capped retries, and an anyio-cancellation-aware sleep.
• No retry for non-idempotent operations: a POST creating a resource is not safely retryable without an idempotency key (applies api/P6 Idempotency).
• except asyncio.CancelledError: raise is the only valid handling — re-raise so the cancellation propagates. Catching and continuing breaks structured concurrency.
• Applies messaging/delivery-semantics: a cancelable async operation is at-most-once; retry-on-cancel is at-least-once. The caller must declare which.

import anyio
import random

async def fetch_retry(id: str, attempts: int = 3) -> User:
    for i in range(attempts):
        try:
            return await fetch_user(id)
        except (TimeoutError, ConnectionError):
            if i == attempts - 1:
                raise
            await anyio.sleep((2 ** i) * 0.1 + random.random() * 0.1)
    raise RuntimeError('unreachable')

Cross-References

• domains/concurrency/patterns.md — the cancellation/timeout/semaphore patterns applied here.
• domains/concurrency/first-principles.md — Concurrency P1 Immutability, P7 Cancellation Support, P8 Timeout Discipline, P9 Bounded Queues.
• domains/messaging/queues.md — asyncio.Queue as an in-process broker; backpressure parallels (IDEATE-40).
• domains/errors/patterns.md — typed async errors and retry-with-backoff.
• languages/py-types.md — Result and exception hierarchy used in async error handling.
• languages/py-tooling.md — pytest-asyncio config that runs these tests.