Python Asyncio In Depth | Event Loop Guide

Introduction

asyncio is Python's standard library for single-threaded concurrency. It lets you manage thousands of connections efficiently for I/O-bound work such as network requests, database queries, and file I/O without threads. asyncio's power comes from understanding the event loop model correctly.

This guide covers how coroutines work, how the event loop schedules tasks, Task and Future abstractions, and batch execution with asyncio.gather. Used incorrectly, asyncio can be slower than threads; with the right patterns it dramatically reduces resource consumption.

Coroutine and async/await Basics

Functions defined with async def return a coroutine object when called; they do not run by themselves. The await expression yields control back to the event loop while waiting for another coroutine to complete. This cooperative multitasking model lets other tasks run while waiting on I/O without blocking the CPU.

You cannot use await inside a synchronous function; you need asyncio.run() or run_until_complete on an existing loop. In mixed codebases, sync_to_async and async_to_sync bridges are common in frameworks like Django and FastAPI.

import asyncio

async def fetch_data(url: str) -> dict:
    await asyncio.sleep(0.1)  # I/O simulation
    return {"url": url, "status": 200}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

Event Loop Architecture

The event loop queues ready coroutines, fires callbacks when I/O completes, and manages timers. asyncio.run() creates a new loop on each call and closes it when done; in long-lived applications keep the loop open and start background work with asyncio.create_task.

get_event_loop() is legacy; prefer asyncio.get_running_loop() from Python 3.10+. When the loop must run blocking CPU or sync I/O in another thread, loop.run_in_executor delegates to a thread pool.

asyncio.run(): ideal for script entry points
create_task(): start background coroutines
run_in_executor(): isolate blocking sync code
call_soon(): schedule callbacks

Task, Future, and gather

asyncio.create_task schedules a coroutine immediately and returns a Task. A Task is a subclass of Future and supports cancel, exception handling, and awaiting results. Use asyncio.gather to run multiple independent I/O operations in parallel.

gather(*aws, return_exceptions=True) lets other tasks finish even if one fails; errors appear as exceptions in the result list. This is critical for bulk API calls where partial failure is acceptable.

async def main():
    tasks = [
        asyncio.create_task(fetch_data(f"/users/{i}"))
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    print(results)

Avoiding Blocking Code

time.sleep(), requests.get(), or synchronous database drivers block the event loop and freeze all async tasks. Solutions: native async libraries like aiohttp, httpx (async mode), asyncpg, aiomysql, or run_in_executor with a thread pool.

CPU-heavy work does not parallelize with asyncio; use multiprocessing or ProcessPoolExecutor instead. Hybrid asyncio + multiprocessing models separate I/O and CPU layers in web servers.

Do not make blocking calls in async code; you freeze the entire event loop.

Timeout, Cancellation, and Error Handling

asyncio.wait_for(coro, timeout=5.0) cancels work that does not finish in time. task.cancel() raises CancellationError; use try/finally inside coroutines for cleanup. asyncio.TaskGroup (3.11+) provides structured concurrency: if one child fails, others are cancelled too.

Exception groups and except* syntax (3.11+) make it easier to handle multiple Task failures in one except block. In production, add timeout and retry policy to every external API call to prevent hung coroutines.

async def with_timeout():
    try:
        return await asyncio.wait_for(fetch_data("/slow"), timeout=2.0)
    except asyncio.TimeoutError:
        return {"error": "timeout"}

Real-World Patterns

FastAPI and Starlette run natively on asyncio via ASGI. Connection pools (asyncpg, redis.asyncio) are shared per loop. For graceful shutdown, cancel tasks in a signal handler and wait for the loop to finish.

Load tests can simulate thousands of concurrent requests with asyncio, but file descriptor limits (ulimit) and pool sizes may bottleneck. Use Semaphore to cap concurrency and prevent resource exhaustion.

Prefer asyncio for I/O-bound work
Use multiprocessing for CPU-bound work
Add timeout to every external call
Limit concurrency with Semaphore

Conclusion

asyncio offers less memory and better scalability than threads for I/O-heavy Python services. Once you internalize the event loop, coroutine, and gather model, you can build production-grade services with FastAPI, aiohttp, and async database drivers.

When migrating blocking code to async, proceed step by step: change the I/O layer first, then convert business logic to coroutines. Profiling how long the loop spends idle clarifies optimization priorities.

Data Usage Notice