Python Async/Await — Write Non-Blocking Code Like a Pro

Introduction

Why This Matters

Python's asyncio module, combined with async/await syntax, allows you to write concurrent code that handles thousands of I/O-bound operations without spawning multiple threads. As APIs, databases, and microservices dominate modern backends, async Python is essential for building high-performance applications.

Without async programming, a web scraper fetching 100 URLs sequentially might take 100 seconds. With asyncio, all 100 requests can run concurrently and complete in under 5 seconds. This performance difference is why frameworks like FastAPI, Starlette, and modern SQLAlchemy now embrace async as a first-class feature.

Understanding async Python also makes you a stronger candidate in technical interviews and a more effective contributor to production codebases that rely on async database drivers like asyncpg, message brokers, and event-driven architectures.

How the Event Loop Works

Python's async system is built around an event loop — a single thread that manages the scheduling of coroutines. Instead of blocking on I/O, coroutines await completion and yield control back to the loop so other tasks can run.

import asyncio
 
async def greet(name: str) -> None:
    await asyncio.sleep(1)  # simulates I/O wait
    print(f"Hello, {name}!")
 
async def main():
    await asyncio.gather(
        greet("Alice"),
        greet("Bob"),
        greet("Charlie"),
    )
 
asyncio.run(main())
# All three greet after ~1 second, not 3 seconds

Coroutines vs Tasks vs Futures

Concept	Description
Coroutine	A function defined with `async def`; not running until awaited
Task	A coroutine wrapped with `asyncio.create_task()` to run concurrently
Future	A low-level object representing a pending result

import asyncio
 
async def fetch_data(id: int) -> str:
    await asyncio.sleep(0.5)
    return f"Data-{id}"
 
async def main():
    # Schedule concurrently with tasks
    task1 = asyncio.create_task(fetch_data(1))
    task2 = asyncio.create_task(fetch_data(2))
 
    result1 = await task1
    result2 = await task2
    print(result1, result2)
 
asyncio.run(main())

Making HTTP Requests with httpx

The httpx library is the async-native replacement for requests.

import asyncio
import httpx
 
async def fetch_url(client: httpx.AsyncClient, url: str) -> str:
    response = await client.get(url)
    return response.text
 
async def main():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3",
    ]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch_url(client, url) for url in urls])
    for r in results:
        print(r[:80])
 
asyncio.run(main())

Async File I/O with aiofiles

import asyncio
import aiofiles
 
async def read_file(path: str) -> str:
    async with aiofiles.open(path, mode="r") as f:
        return await f.read()
 
async def write_file(path: str, content: str) -> None:
    async with aiofiles.open(path, mode="w") as f:
        await f.write(content)
 
async def main():
    await write_file("/tmp/test.txt", "Hello async world!")
    content = await read_file("/tmp/test.txt")
    print(content)
 
asyncio.run(main())

Timeouts and Error Handling

import asyncio
import httpx
 
async def fetch_with_timeout(url: str) -> str:
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(url)
            response.raise_for_status()
            return response.text
    except httpx.TimeoutException:
        return "Request timed out"
    except httpx.HTTPStatusError as e:
        return f"HTTP error: {e.response.status_code}"
 
async def main():
    result = await fetch_with_timeout("https://httpbin.org/delay/1")
    print(result[:100])
 
asyncio.run(main())

asyncio.gather vs asyncio.wait

import asyncio
 
async def task(n: int) -> int:
    await asyncio.sleep(n * 0.1)
    return n * 2
 
async def main():
    # gather: returns results in order, raises on first exception
    results = await asyncio.gather(task(1), task(2), task(3))
    print(results)  # [2, 4, 6]
 
    # wait: returns sets of done/pending tasks
    tasks = [asyncio.create_task(task(i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(tasks, timeout=0.25)
    print(f"Done: {len(done)}, Pending: {len(pending)}")
 
asyncio.run(main())

Common Mistakes

Calling a coroutine without await — it returns a coroutine object, not the result
Using time.sleep() instead of asyncio.sleep() — blocks the entire event loop
Running CPU-bound work inside async functions — use asyncio.run_in_executor() instead
Forgetting to use async with for async context managers like httpx.AsyncClient
Creating tasks but not awaiting them — they may be garbage collected before completion

Best Practices

Use asyncio.run() as your entry point, not loop.run_until_complete()
Prefer asyncio.gather() for concurrent I/O; use asyncio.create_task() for fire-and-forget
Use anyio or trio for more structured concurrency in complex applications
Add timeouts to all external calls to prevent hanging tasks
Profile with asyncio debug mode: PYTHONASYNCIODEBUG=1 python script.py

Key Takeaways

async def defines a coroutine; await suspends it until the result is ready
The asyncio event loop runs in a single thread and schedules coroutines cooperatively
asyncio.gather() runs multiple coroutines concurrently and collects results in order
httpx.AsyncClient and aiofiles are the go-to libraries for async HTTP and file I/O
CPU-bound tasks should use ProcessPoolExecutor via run_in_executor, not raw async
Python 3.11+ introduced asyncio.TaskGroup for structured concurrency with better error handling
Async Python is the foundation of FastAPI, modern SQLAlchemy, and event-driven microservices

Introduction#

Why This Matters#

How the Event Loop Works#

Coroutines vs Tasks vs Futures#

Making HTTP Requests with httpx#

Async File I/O with aiofiles#

Timeouts and Error Handling#

asyncio.gather vs asyncio.wait#

Common Mistakes#

Best Practices#

Key Takeaways#

Related reading