Async Programming in Python with asyncio

Async Programming in Python with asyncio

asyncio is Python's standard library for cooperative I/O concurrency. It lets you run thousands of network operations simultaneously in a single thread — no GIL contention, no OS thread overhead.

Core concepts

Getting started

import asyncio

async def greet(name: str) -> str:
    await asyncio.sleep(1)   # non-blocking sleep — yields control
    return f"Hello, {name}!"

async def main():
    result = await greet("world")
    print(result)

asyncio.run(main())   # creates, runs, and closes the event loop

Concurrency with Tasks and gather

import asyncio
import time

async def fetch_mock(url: str, delay: float) -> dict:
    print(f"  → starting {url}")
    await asyncio.sleep(delay)
    print(f"  ✓ done {url} ({delay:.1f}s)")
    return {"url": url, "status": 200}

async def sequential():
    t0 = time.perf_counter()
    for url, d in [("api/users", 0.8), ("api/posts", 1.2), ("api/comments", 0.5)]:
        await fetch_mock(url, d)
    print(f"Sequential: {time.perf_counter()-t0:.2f}s")

async def concurrent():
    t0 = time.perf_counter()
    results = await asyncio.gather(
        fetch_mock("api/users",    0.8),
        fetch_mock("api/posts",    1.2),
        fetch_mock("api/comments", 0.5),
    )
    print(f"Concurrent: {time.perf_counter()-t0:.2f}s")
    return results

asyncio.run(sequential())   # ~2.5s
asyncio.run(concurrent())   # ~1.2s  (bottleneck is the slowest)

Tasks — granular control

async def task_with_progress(name: str, steps: int):
    for i in range(1, steps + 1):
        await asyncio.sleep(0.1)
        print(f"{name}: step {i}/{steps}")
    return f"{name} done"

async def main():
    task_a = asyncio.create_task(task_with_progress("A", 3), name="task-A")
    task_b = asyncio.create_task(task_with_progress("B", 5), name="task-B")

    result_a = await task_a
    print(result_a)

    result_b = await task_b
    print(result_b)

asyncio.run(main())

gather vs TaskGroup (Python 3.11+)

# gather — configurable error handling
results = await asyncio.gather(
    coro1(), coro2(), coro3(),
    return_exceptions=True,  # returns Exception objects instead of raising
)
successes = [r for r in results if not isinstance(r, Exception)]
failures  = [r for r in results if isinstance(r, Exception)]

# TaskGroup — cancels all tasks if any fails (safer, Python 3.11+)
async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(coro1())
        t2 = tg.create_task(coro2())
    # both tasks completed here (or ExceptionGroup was raised)
    print(t1.result(), t2.result())

Timeouts

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "data"

async def main():
    # Python 3.11+ — asyncio.timeout context manager
    try:
        async with asyncio.timeout(2.0):
            result = await slow_operation()
    except TimeoutError:
        print("Operation cancelled — timeout exceeded")

    # Python 3.9+ compatible
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Timeout (wait_for)")

asyncio.run(main())

Semaphores and concurrency control

import asyncio
import aiohttp  # pip install aiohttp

MAX_CONCURRENT = 10

async def fetch_url(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> dict:
    async with sem:   # at most MAX_CONCURRENT simultaneous requests
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            data = await resp.json()
            return {"url": url, "status": resp.status, "data": data}

async def fetch_many(urls: list[str]) -> list[dict]:
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    async with aiohttp.ClientSession() as session:
        tasks   = [fetch_url(session, url, sem) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if isinstance(r, dict)]
    errors    = [r for r in results if isinstance(r, Exception)]
    print(f"OK: {len(successes)} | Errors: {len(errors)}")
    return successes

urls    = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 51)]
results = asyncio.run(fetch_many(urls))

Producer-consumer pattern with Queue

import asyncio
import random

async def producer(queue: asyncio.Queue, n: int, pid: int):
    for i in range(n):
        item = f"item-{pid}-{i}"
        await queue.put(item)
        print(f"[P{pid}] produced {item} (queue: {queue.qsize()})")
        await asyncio.sleep(random.uniform(0.05, 0.15))
    print(f"[P{pid}] done")

async def consumer(queue: asyncio.Queue, cid: int):
    while True:
        try:
            item = await asyncio.wait_for(queue.get(), timeout=1.0)
            print(f"  [C{cid}] processing {item}")
            await asyncio.sleep(random.uniform(0.1, 0.3))
            queue.task_done()
        except asyncio.TimeoutError:
            print(f"  [C{cid}] idle — exiting")
            break

async def pipeline():
    queue = asyncio.Queue(maxsize=20)
    async with asyncio.TaskGroup() as tg:
        for i in range(3):
            tg.create_task(producer(queue, n=5, pid=i))
        for j in range(4):
            tg.create_task(consumer(queue, cid=j))

asyncio.run(pipeline())

Async HTTP API client with aiohttp

import asyncio
import aiohttp

async def paginated_client(base_url: str) -> list:
    results = []
    page    = 1

    async with aiohttp.ClientSession(
        headers={"Accept": "application/json"},
        timeout=aiohttp.ClientTimeout(total=30),
    ) as session:
        while True:
            async with session.get(f"{base_url}?_page={page}&_limit=10") as resp:
                if resp.status != 200:
                    break
                data = await resp.json()
                if not data:
                    break
                results.extend(data)
                page += 1

    return results

async def main():
    posts = await paginated_client("https://jsonplaceholder.typicode.com/posts")
    print(f"Total posts downloaded: {len(posts)}")

asyncio.run(main())

Running blocking code without blocking the loop

import asyncio
import time

def cpu_bound(n: int) -> int:
    """Simulates a heavy CPU computation."""
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()

    # run_in_executor — default ThreadPoolExecutor
    result = await loop.run_in_executor(None, cpu_bound, 10_000_000)
    print(f"Result: {result}")

    # asyncio.to_thread (Python 3.9+, cleaner)
    result2 = await asyncio.to_thread(time.sleep, 0.5)  # non-blocking sleep via thread
    print("Thread done")

asyncio.run(main())

Debugging tips

# Enable debug mode — detects unawaited coroutines and slow callbacks
asyncio.run(main(), debug=True)

# Inspect running tasks from inside a coroutine
async def inspect():
    for task in asyncio.all_tasks():
        print(f"  {task.get_name()!r} — done={task.done()}")

Best practices

• asyncio.run(main()) is the correct entry point in scripts — don't call loop.run_until_complete() manually (legacy API).
• Never call blocking code (time.sleep, sync file I/O, CPU-heavy loops) inside a coroutine — it stalls the entire event loop. Use asyncio.to_thread() or run_in_executor() to offload to a thread pool.
• gather(return_exceptions=True) for resilient fan-out — separate dict results from Exception instances after awaiting.
• Prefer asyncio.TaskGroup (3.11+) when a failure in any subtask should cancel the rest — it's safer than unguarded gather.
• Use asyncio.Semaphore when calling rate-limited APIs or when you need to cap memory usage from too many concurrent downloads.
• Always use async with aiohttp.ClientSession() — it ensures connections are properly closed and avoids resource-leak warnings.

Related conversions

Frequent conversions across the catalogue:

• JPG to PNG
• MP4 to MP3
• PDF to DOCX
• PNG to JPG
• MP3 to WAV
• DOCX to PDF