Python Async Patterns: asyncio Without the Footguns

Python's asyncio is powerful and full of footguns. Common patterns from synchronous code — sleep, queue, lock — all have async equivalents, but the semantics are different enough that "I'll just put async/await everywhere" produces deadlocks, leaked tasks, and resources that never close.

This post is the patterns that actually work in production asyncio code, the ones that look correct but bite, and the asyncio features that arrived in Python 3.11+ that you should be using.

TaskGroup Over Gather

Pre-3.11 asyncio code is full of asyncio.gather(). It works, but it has a problem: if one of the awaitables raises, the others keep running, and their exceptions are silently swallowed unless you also pass return_exceptions=True.

asyncio.TaskGroup (3.11+) fixes this:

import asyncio

async def fetch_all(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]
    return [t.result() for t in tasks]

If any task raises, the TaskGroup cancels the others, waits for them to clean up, and re-raises an ExceptionGroup with all the errors. Cleanup is deterministic. No leaked tasks.

Use TaskGroups for any code that spawns a known set of concurrent tasks. Reserve raw gather for cases where you specifically want to collect partial results.

Cancellation Is Cooperative

Async tasks do not get killed externally. Cancellation is a request — a CancelledError raised at the next await. Code that does not check for it can ignore the request.

async def long_running():
    for i in range(1_000_000):
        process(i)  # synchronous — no cancellation check
    # CancelledError might wait minutes

Two failure modes follow:

• Synchronous-heavy code is not cancellable. Long synchronous computation in an async function blocks the event loop and ignores cancellation.
• Catching CancelledError too broadly silences cancellation. except Exception does not catch CancelledError in 3.8+, but except BaseException does. If you catch and ignore, the task never dies.

The pattern: yield to the event loop periodically with await asyncio.sleep(0) in long-running synchronous loops, or run them in an executor.

async def long_running():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, sync_heavy_work)

Timeouts With asyncio.timeout

Pre-3.11: asyncio.wait_for. Post-3.11: asyncio.timeout. The new version composes better and handles cancellation more cleanly.

async def fetch_with_timeout(url):
    async with asyncio.timeout(5.0):
        return await fetch(url)

If the operation takes longer than 5 seconds, the inner code is cancelled and a TimeoutError is raised at the async with line. Cleanup happens through normal context manager semantics.

Wrap external calls in timeouts as a default. An unbounded await is a guaranteed production incident eventually.

Locks and Semaphores

asyncio locks look like threading locks but behave differently. Most importantly: they are cooperative, single-threaded, and only meaningful inside one event loop.

import asyncio

# Limit concurrent API calls
semaphore = asyncio.Semaphore(10)

async def call_api(url):
    async with semaphore:
        return await client.get(url)

A semaphore is the right tool when you have many concurrent tasks and want to bound how many run at once. The pattern: launch all tasks, let the semaphore admit them in batches.

Locks (asyncio.Lock) are mostly useful for protecting shared state from interleaved access. In single-threaded async code, race conditions still exist — they just happen at await points, not between threads.

balance = 0
lock = asyncio.Lock()

async def transfer(amount):
    async with lock:
        global balance
        current = balance
        await asyncio.sleep(0)  # yields to other tasks
        balance = current + amount  # safe because we hold the lock

Without the lock, two concurrent transfer calls can interleave at the await and produce wrong results.

Sync/Async Boundaries

The classic mistake: calling a blocking sync function from async code without using an executor.

import requests  # sync HTTP client

async def fetch(url):
    return requests.get(url)  # blocks the entire event loop

This blocks every async task in your process until requests.get returns. Under load, the event loop is unresponsive.

Fix one of two ways:

# Option 1: use an async HTTP client
import httpx
async with httpx.AsyncClient() as client:
    return await client.get(url)

# Option 2: run the sync call in an executor
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, requests.get, url)

Async code that does sync I/O is a leaky abstraction. Most application code should either be fully async (with async libraries throughout) or fully sync. Mixed code is the unhappy middle.

Async Generators

Async generators are powerful for streaming. The syntax is straightforward:

async def fetch_pages(url):
    page = 1
    while True:
        resp = await client.get(f"{url}?page={page}")
        if not resp.json()["items"]:
            return
        for item in resp.json()["items"]:
            yield item
        page += 1

async def main():
    async for item in fetch_pages("https://api.example.com/items"):
        process(item)

The gotcha: async generators have lifecycle requirements. If the generator is garbage collected with pending state, asyncio may complain or leak resources. Use them inside async for loops or aclose() them explicitly.

Common Anti-Patterns

Fire-and-forget tasks without TaskGroup. Creating a task with asyncio.create_task() and not awaiting it. If the task raises, the exception goes nowhere. If the program exits, the task may be cancelled mid-work.

# Bad
asyncio.create_task(send_notification(user))

# Better — track the task and await it
task = asyncio.create_task(send_notification(user))
# ... eventually
await task

# Best — use a TaskGroup if multiple
async with asyncio.TaskGroup() as tg:
    tg.create_task(send_notification(user))

Mixing sync and async I/O in the same function. Confuses callers and makes timing unpredictable. Pick one.

Spawning unbounded tasks. "I'll process each message in a new task." If messages arrive faster than tasks complete, you have unbounded memory growth. Use a semaphore.

asyncio.run() inside another async function. Causes RuntimeError. There can be only one event loop per thread. Use await for nested async, not asyncio.run().

Testing Async Code

Use pytest-asyncio or anyio:

import pytest

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch("https://example.com")
    assert result.status_code == 200

For testing concurrency, prefer to test deterministically by injecting fakes that complete in known orders. Real network timing in tests produces flaky tests.

When Async Pays Off

Async Python is best for:

• I/O-bound workloads with many concurrent connections (web servers, message processors)
• Crawlers, scrapers, batch API consumers
• Long-lived connections (WebSockets, server-sent events)

Async does not help with:

• CPU-bound work (use multiprocessing or a real concurrent language)
• Heavily sync ecosystems (Django pre-async, sqlite without aiosqlite)
• Code that has to call legacy sync libraries

If your codebase is async and you find yourself wrapping everything in run_in_executor, you are probably better off being sync. Async is a commitment.

Reviewing an asyncio codebase that has accumulated subtle bugs and resource leaks? We help teams audit for cancellation correctness, sync/async boundaries, and leaked tasks. scopeforged.com