3 Async Python Patterns I Wish I Learned Sooner (With Real Code)

I spent two years writing async Python wrong. Not "my code crashed" wrong — more like "I was leaving 70% of the performance on the floor and didn't know it" wrong.

Here are three patterns that actually changed how I write async code. Each one came from a real production problem that forced me to dig deeper than the basic async/await syntax everyone learns on day one.

Pattern 1: asyncio.gather — Stop Awaiting Things One at a Time

The mistake I kept making for way too long: chaining await calls sequentially even when they had no reason to wait for each other.

# What I used to write (embarrassing in hindsight):
user = await db.fetch_user(user_id)
orders = await db.fetch_orders(user_id)
notifs = await db.fetch_notifications(user_id)
# Total: fetch_user + fetch_orders + fetch_notifications = ~600ms

Each await blocks the coroutine until that specific call finishes. If each database query takes ~200ms, you're looking at 600ms total. But here's the thing — the database can handle all three queries at once. I was the bottleneck, not the database.

# What I write now:
user, orders, notifs = await asyncio.gather(
    db.fetch_user(user_id),
    db.fetch_orders(user_id),
    db.fetch_notifications(user_id),
)
# Total: max(200ms, 200ms, 200ms) = ~200ms

asyncio.gather schedules all the coroutines at once. It returns only when the slowest one finishes. Three independent I/O calls that took 600ms now take 200ms. That's a 3x speedup from one line change.

Real numbers from a project dashboard endpoint I optimized: 1.2 seconds → 340 milliseconds. Three database queries and one Redis cache lookup all running concurrently instead of one after another. The user noticed — 340ms feels instant, 1.2s feels sluggish.

One gotcha: exceptions inside gather can be tricky. By default, the first exception raised cancels all other tasks and propagates up. If you want to collect all results (including failures), use return_exceptions=True:

results = await asyncio.gather(
    fetch_a(), fetch_b(), fetch_c(),
    return_exceptions=True
)
for i, r in enumerate(results):
    if isinstance(r, Exception):
        logger.warning(f"Source {i} failed: {r}")
    else:
        process(r)

This pattern alone probably saved me more latency than every other optimization combined.

Pattern 2: asyncio.Semaphore — The Bouncer Your API Calls Need

I learned this one the hard way. I had a FastAPI endpoint that needed to enrich results by calling a third-party API. Simple enough — fire off 100 concurrent requests, wait for all of them, and return.

What actually happened: the third-party API had a rate limit of 10 requests per second. My endpoint sent all 100 in the first 100 milliseconds. They rate-limited me. My endpoint returned errors. Users were not happy.

Enter asyncio.Semaphore. Think of it as a bouncer at a club — only N people get in at a time:

import asyncio

async def fetch_with_limit(urls: list[str], max_concurrent: int = 5):
    sem = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url):
        async with sem:
            # Only `max_concurrent` coroutines run this block at once
            resp = await http_client.get(url)
            return resp.json()

    tasks = [fetch_one(url) for url in urls]
    return await asyncio.gather(*tasks)

# 100 URLs, never more than 5 concurrent HTTP calls:
results = await fetch_with_limit(all_urls, max_concurrent=5)

The async with sem acquires a slot. If all 5 are taken, the 6th coroutine just pauses there — no error, no polling, no busy-waiting. It resumes automatically when a slot frees up.

I use semaphores everywhere now: database connection pools, file descriptor limits, rate-limited APIs. It's 6 lines of code (the sem = line and the async with sem: block) that prevent entire categories of production incidents.

Pro tip: combine with exponential backoff for extra resilience:

async def fetch_with_retry(url, sem, max_retries=3):
    for attempt in range(max_retries):
        try:
            async with sem:
                return await http_client.get(url)
        except RateLimitError:
            wait = 2 ** attempt  # 1s → 2s → 4s
            await asyncio.sleep(wait)
    raise Exception(f"Failed after {max_retries} retries: {url}")

Pattern 3: asyncio.Queue — Backpressure Without the Headache

The hardest bug I've debugged in async Python: memory slowly creeping up over hours until the process got OOM-killed at 2 AM. No crash, no traceback, just a container restart and confused morning investigation.

The culprit was deceptively simple — a list acting as an unbounded buffer between a fast producer and a slow consumer:

# The problem (memory leak in disguise):
results = []
async for item in fast_source():  # Produces 10,000 items/second
    results.append(item)          # List grows without bound

for item in results:              # Consumer processes at 100 items/second
    await slow_process(item)      # 10,000 items sitting in RAM, growing every second

The producer was an event stream that never stopped. The consumer was a database writer that could only handle ~100 inserts per second. The gap was filled by RAM — until RAM ran out.

asyncio.Queue with a maxsize fixes this by applying natural backpressure:

import asyncio

async def pipeline():
    queue = asyncio.Queue(maxsize=50)  # Never more than 50 items in memory

    async def producer():
        async for item in fast_source():
            await queue.put(item)  # BLOCKS if queue is full — backpressure!
        await queue.put(None)      # Sentinel value signals "we're done"

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:       # Sentinel received → exit cleanly
                break
            await slow_process(item)
            queue.task_done()

    await asyncio.gather(producer(), consumer())

The magic is queue.put(item) blocking when the queue is full. The producer automatically slows down to the consumer's speed. Memory usage stays flat at maxsize items. No monitoring alerts, no OOM-kills, no 2 AM wake-up calls.

I now use this pattern for:

• Streaming large files through a transform pipeline (read → parse → write)
• Batch-inserting database records (consumer collects 100 items, flushes in one INSERT)
• Web scraping with download → parse → store stages

The maxsize number matters. Too small and you underutilize the consumer. Too large and you waste memory. My rule of thumb: start with maxsize = consumer_items_per_second * 2 and tune from there.

What I'd Tell My Past Self

1. Every sequential await is a question. "Does this actually depend on the previous call?" If not → gather it.
2. Every external API call gets a semaphore. It's 6 lines of code that prevent midnight incidents.
3. Every pipeline with different speeds gets a queue with maxsize. Backpressure is automatic — you just have to set it up.

None of these are advanced async patterns. They're basic building blocks that the async/await tutorial everyone reads somehow skips. I missed them for two years. Don't make the same mistake.

---

Found a pattern that changed how you write async code? I'm always looking for the next thing I should have learned two years ago.