Master Python asyncio: Make Your Code Fly with Asynchronous Programming

Why synchronous code stalls

When a Python program waits for network or file I/O, the CPU sits idle like a cashier waiting for a customer to finish paying. The article starts with this analogy and shows a simple web‑crawler scenario where each request blocks the next.

Async programming with asyncio

asyncio

enables cooperative multitasking: the main thread can switch from one coroutine to another while the former is awaiting I/O. This is illustrated with a grocery‑store analogy comparing synchronous (one‑cashier) and asynchronous (cashier can serve other customers while one pays).

Synchronous vs asynchronous example

# Synchronous version – runs three tasks sequentially, takes ~3 s
import time

def sync_task(n):
    print(f"Task {n} start")
    time.sleep(1)  # simulate I/O
    print(f"Task {n} end")
    return n

start = time.time()
for i in range(1, 4):
    sync_task(i)
print(f"Sync elapsed: {time.time() - start:.2f}s")

# Asynchronous version – runs three tasks concurrently, takes ~1 s
import asyncio

async def async_task(n):
    print(f"Async task {n} start")
    await asyncio.sleep(1)
    print(f"Async task {n} end")
    return n

async def main():
    start = time.time()
    tasks = [async_task(i) for i in range(1, 4)]
    results = await asyncio.gather(*tasks)
    print(f"Async elapsed: {time.time() - start:.2f}s")
    print(f"Results: {results}")

asyncio.run(main())

The output shows the asynchronous version finishes in about one second, demonstrating the performance gain.

Core concepts: Event loop and coroutine

A coroutine is defined with async def and can be paused with await. The event loop continuously checks which coroutines are ready to run, schedules them, and resumes them after their awaitable completes.

Creating and running coroutines

import asyncio

async def fetch_data(url):
    print(f"Start fetching: {url}")
    await asyncio.sleep(2)  # simulate network delay
    print(f"Finished fetching: {url}")
    return f"Data from {url}"

asyncio.run(fetch_data('https://example.com'))

Task management

Coroutines become runnable tasks when wrapped in asyncio.create_task() or placed in a TaskGroup (Python 3.11+). Tasks are scheduled by the event loop.

Creating tasks with create_task

import asyncio

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    task1 = asyncio.create_task(say_after(1, 'Hello'))
    task2 = asyncio.create_task(say_after(2, 'World'))
    print('Tasks created, main continues...')
    await task1
    await task2
    print('All tasks done!')

asyncio.run(main())

Important note: if a task is created without keeping a reference or awaiting it, the task may be garbage‑collected and never run.

Task groups for safer concurrency

import asyncio

async def worker(name, seconds):
    await asyncio.sleep(seconds)
    print(f"{name} completed")
    return f"{name}-result"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(worker('Task1', 1))
            t2 = tg.create_task(worker('Task2', 2))
    except* Exception as eg:
        print(f"Error: {eg.exceptions}")
    else:
        print(f"All succeeded: {t1.result()}, {t2.result()}")

asyncio.run(main())

TaskGroup automatically cancels remaining tasks if any task raises an exception, preventing resource leaks.

Concurrent execution patterns

The article compares three high‑level helpers: asyncio.gather() – waits for all tasks and returns results in the order they were passed. asyncio.as_completed() – yields futures as they finish, allowing processing in completion order. asyncio.wait() – returns two sets (done, pending) and can stop after the first task finishes.

# Example using gather
async def demo_gather():
    results = await asyncio.gather(task_a(), task_b(), task_c())
    print(results)

# Example using as_completed
async def demo_as_completed():
    tasks = [task_x(), task_y(), task_z()]
    for fut in asyncio.as_completed(tasks):
        result = await fut
        print('Got:', result)

# Example using wait with timeout
async def demo_wait():
    tasks = [worker(i) for i in range(1,5)]
    done, pending = await asyncio.wait(tasks, timeout=2.5, return_when=asyncio.FIRST_COMPLETED)
    print(f"Completed: {len(done)}; still pending: {len(pending)}")

Advanced tips and pitfalls

Timeout control

import asyncio

async def eternity():
    await asyncio.sleep(3600)
    print('This will never be seen')

async def main():
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except asyncio.TimeoutError:
        print('Task timed out and was cancelled')

asyncio.run(main())

Python 3.11 also provides asyncio.timeout() as a context manager.

Shielding critical operations

async def critical_operation():
    try:
        await asyncio.sleep(2)
        print('Critical work done')
        return 'important data'
    except asyncio.CancelledError:
        print('Cancelled but cleaning up')
        await asyncio.sleep(1)
        print('Cleanup finished')
        raise

async def main():
    task = asyncio.create_task(critical_operation())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await asyncio.shield(task)
    except asyncio.CancelledError:
        print('Main saw cancellation, but task continues in background')
    await asyncio.sleep(2.5)  # give the task time to finish

asyncio.run(main())

Running blocking code without freezing the loop

import asyncio, time

def blocking_io():
    time.sleep(2)
    return 'result from blocking call'

async def main():
    print('Start async part')
    await asyncio.sleep(1)
    result = await asyncio.to_thread(blocking_io)
    print('Got:', result)
    print('Continue with other async work')

asyncio.run(main())

Common pitfalls

Calling a blocking function directly inside an async function blocks the whole loop.

Forgetting await on a coroutine means it never runs.

Creating too many tasks at once can exhaust resources; use Semaphore or connection pools to limit concurrency.

Best‑practice checklist

Always use async/await syntax; avoid manual loop management.

Control concurrency with Semaphore or a limited connector pool.

Set timeouts for every async operation to prevent hangs.

Wrap await calls in try/except to handle exceptions gracefully.

Never call blocking code directly; use to_thread or a process pool.

Keep references to created tasks so they are not garbage‑collected.

Prefer TaskGroup (Python 3.11+) for related tasks to get automatic cancellation and clearer error handling.

Conclusion

asyncio

opens a new door for Python concurrency by providing a single‑threaded, cooperative multitasking model that handles thousands of I/O‑bound connections with minimal overhead. Mastering coroutines, the event loop, task creation, timeout handling, and the patterns above equips you to build high‑performance network services, crawlers, and real‑time applications.