Running I/O-Bound Work Concurrently with Python asyncio: A Task Guide

Desired Outcome

Execute multiple I/O-bound operations — HTTP requests, database queries, file reads, or external API calls — concurrently on a single thread so that total wall-clock time approaches the slowest individual operation rather than the sum of all operations. The event loop manages readiness notifications; your coroutines yield at await points, allowing other tasks to progress while one waits for the network or disk.

Prerequisites

• Python 3.11 or later (behavior and defaults differ on earlier versions, especially Windows event-loop policy).
• All I/O libraries used must be async-native (e.g., aiohttp, asyncpg, aiofiles) or wrapped via asyncio.to_thread() / run_in_executor(). Blocking calls like requests.get() or time.sleep() inside a coroutine stall the entire loop.
• Entry point is a synchronous script or REPL; you will call asyncio.run(main()) once at the top level.

Focused Procedure

1. Define coroutines for each independent I/O operation

Each coroutine performs one logical unit of work and returns a result or raises an exception. Keep them small and focused.

import asyncio
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

2. Schedule concurrent work with asyncio.create_task() or asyncio.gather()

Use create_task() when you need to start work early and collect results later (fire-and-forget with later await). Use gather() when you want to wait for all results in input order.

async def main() -> list[dict]:
    urls = [
        "https://api.example.com/item/1",
        "https://api.example.com/item/2",
        "https://api.example.com/item/3",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

Decision point: return_exceptions=True makes gather return exceptions as values instead of raising the first one. Choose this when partial success is acceptable and you want to inspect each outcome individually. With the default False, the first raised exception propagates immediately and sibling tasks are cancelled.

3. Bound external calls with timeouts

Wrap any awaitable that talks to the network in asyncio.wait_for() so a hung connection fails predictably instead of pinning the loop forever.

async def fetch_with_timeout(session: aiohttp.ClientSession, url: str, timeout: float = 5.0) -> dict:
    try:
        return await asyncio.wait_for(fetch(session, url), timeout=timeout)
    except asyncio.TimeoutError:
        raise RuntimeError(f"Request to {url} exceeded {timeout}s")

4. Offload blocking or CPU-bound work

If you must call a synchronous library (e.g., legacy requests, heavy JSON parsing, cryptographic operations), run it in a thread pool so the event loop stays free.

import json

def heavy_parse(payload: bytes) -> dict:
    # Simulates CPU-bound work
    return json.loads(payload)

async def fetch_and_parse(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        data = await resp.read()
    # Offload to default thread pool executor
    return await asyncio.to_thread(heavy_parse, data)

For sustained CPU-heavy workloads, use ProcessPoolExecutor via loop.run_in_executor() to bypass the GIL entirely.

5. Clean shutdown

When main() completes or is cancelled, asyncio.run() cancels all pending tasks. Await them with return_exceptions=True so cleanup finally blocks run without unhandled-exception noise. Close async clients (sessions, pools) before the loop exits.

async def main() -> list[dict]:
    urls = [...]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch_with_timeout(session, u)) for u in urls]
        try:
            return await asyncio.gather(*tasks, return_exceptions=True)
        finally:
            # Ensure all tasks are awaited even if gather raises
            for t in tasks:
                if not t.done():
                    t.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)

Expected Checks

Concurrency verification

Run two coroutines that each await asyncio.sleep(1). Total wall time should be ~1 second, not 2.

async def test_concurrency():
    async def sleeper():
        await asyncio.sleep(1)
    start = asyncio.get_event_loop().time()
    await asyncio.gather(sleeper(), sleeper())
    elapsed = asyncio.get_event_loop().time() - start
    assert elapsed < 1.5, f"Expected ~1s, got {elapsed:.2f}s"

Run this with python -m pytest test_concurrency.py -v (requires pytest). Permission: standard user. Risk: none.

Partial-failure behavior

Add one failing coroutine and compare gather default vs return_exceptions=True.

async def failer():
    raise ValueError("boom")

# Default: raises ValueError, other tasks cancelled
# return_exceptions=True: returns [result1, ValueError("boom"), result3]

Timeout enforcement

Wrap a deliberately slow awaitable in asyncio.wait_for(..., timeout=0.1) and confirm TimeoutError is raised.

async def slow():
    await asyncio.sleep(10)

try:
    await asyncio.wait_for(slow(), timeout=0.1)
except asyncio.TimeoutError:
    print("timeout works")

Cancellation cleanup

Create then cancel a task, await it, and verify finally executes without unhandled-exception noise.

async def with_cleanup():
    try:
        await asyncio.sleep(10)
    finally:
        print("cleanup ran")

task = asyncio.create_task(with_cleanup())
task.cancel()
await asyncio.gather(task, return_exceptions=True)
# Output should show "cleanup ran" and no traceback

Blocking-call detection

Introduce one time.sleep(1) inside a coroutine alongside two asyncio.sleep(1) coroutines. Observe that all three serialize (~3s total). Replace time.sleep with await asyncio.to_thread(time.sleep, 1) and re-check timing (~1s).

Recovery Options

Task cancellation

Cancellation is cooperative: the cancelled task receives CancelledError at the next await. If your code catches Exception broadly, re-raise CancelledError or use except* (Python 3.11+) to avoid swallowing it. Cleanup in finally blocks runs before the task terminates.

Exception handling in gather

With return_exceptions=True, iterate results and handle each BaseException subclass individually. Log, retry, or substitute a fallback value per item.

Loop shutdown warnings

If you see Task was destroyed but it is pending! or unclosed client session warnings, ensure you await all tasks in a finally block and close async context managers (async with or explicit await session.close()) before asyncio.run() returns.

Limitations

• asyncio provides concurrency, not parallelism. CPU-bound work blocks the loop; use to_thread() or ProcessPoolExecutor.
• Windows defaults to ProactorEventLoop (Python 3.8+); some libraries expect SelectorEventLoop. Set policy explicitly if needed: asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) before asyncio.run().
• Third-party async libraries may have their own connection-pool limits; tune aiohttp.TCPConnector(limit=...) to avoid overwhelming the target service.

Practical Verification Checklist

1. Run the concurrency test; confirm ~1s wall time for two 1-second sleeps.
2. Run the timeout test; confirm TimeoutError at 0.1s.
3. Run the cancellation test; confirm "cleanup ran" prints with no traceback.
4. Run the blocking-call test; confirm serialization without to_thread and concurrency with it.
5. In your real workload, log start/end timestamps per task and verify total time matches the slowest task, not the sum.