Python Async

Python Asyncio

Standards for concurrency with asyncio.

Run independent work concurrently

• Never await independent coroutines one at a time in a loop — that runs them serially

and is a performance red flag. Schedule them together so they overlap.

• Use asyncio.TaskGroup for fail-fast semantics — the first error cancels its siblings.

Plain asyncio.gather lets the siblings run on (the first exception still propagates), while gather(..., return_exceptions=True) returns every result, exceptions included. The choice is about error propagation — document it when it isn't obvious.

• Bound fan-out with an asyncio.Semaphore: unbounded concurrency usually exhausts host

memory (the more common failure) and can open too many connections to a downstream service. Expose the limit as a service setting so it can be tuned without a new release.

• Sequentially awaiting paginated pages (each request needs the previous page's cursor)

is the legitimate exception — keep it rare and flag it with a comment.

import asyncio
from collections.abc import Sequence

async def fetch_all(record_ids: Sequence[str]) -> list[Record]:
    semaphore = asyncio.Semaphore(10)

    async def fetch_one(record_id: str) -> Record:
        async with semaphore:
            return await client.get(record_id)

    async with asyncio.TaskGroup() as task_group:
        tasks = [task_group.create_task(fetch_one(rid)) for rid in record_ids]
    return [task.result() for task in tasks]

Async all the way down

• In FastAPI, mark an endpoint async def only if it awaits real async I/O (an async DB

driver) — not if it makes blocking calls. A blocking call inside an async endpoint stalls the event loop; either make it genuinely async or leave the endpoint a plain def so the framework runs it in a threadpool.

• Don't mark a function async def if it never awaits — it gains no concurrency, misleads

readers, and can be marginally slower.

• Create async clients, sessions, and connection pools once at startup and inject them;

constructing them is costly enough to outweigh the gain from concurrency if done per request.

• Deploy asyncio services on uvloop, a faster drop-in event loop; the built-in loop is

fine for local runs and debugging but slower in production.

Errors from concurrent tasks

• asyncio.TaskGroup raises an ExceptionGroup on failure and cancels siblings — prefer

it when you don't need to salvage the results of the siblings it cancels.

• With gather(..., return_exceptions=True), inspect every result: immediately re-raise

anything that is a BaseException but not an Exception (asyncio.CancelledError, SystemExit), and collect the plain Exceptions into an ExceptionGroup to raise together. Never swallow a BaseException with a bare continue.

• Bound polling and retry loops with a max iteration count or an asyncio.timeout, so a

loop can't run forever when its condition never becomes true.

See python-error-handling for catching the narrowest type and custom exceptions.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: owenlamont
• Source: owenlamont/agent_skills
• License: MIT

Install and usage instructions live in the source repository linked above.