Python Async/Await Complete Guide: asyncio, aiohttp and Concurrency



Python Async/Await: The Complete Guide to Asynchronous Programming

Master concurrency in Python — write non-blocking I/O code that's 10–100× faster than synchronous alternatives.

Contents

1. Why Async? Sync vs Async vs Threads
2. The Event Loop Explained
3. Coroutines, async/await Syntax
4. Tasks and Gathering
5. Real I/O: aiohttp, aiofiles, asyncpg
6. Production Patterns
7. Common Pitfalls
8. Benchmark: Sync vs Async

Why Async? Sync vs Async vs Threads

Python's GIL prevents true parallel CPU execution in threads, but I/O-bound work (HTTP requests, DB queries, file reads) spends most of its time waiting. Async lets you do something else during that wait.

Approach	Best For	Drawback
Synchronous	Simple scripts, CPU-bound	Blocks on every I/O call
Threading	I/O-bound, legacy code	GIL, memory overhead, race conditions
Multiprocessing	CPU-bound tasks	High memory, IPC complexity
Async/Await	I/O-bound, high concurrency	Requires async-aware libraries

The Event Loop Explained

The event loop is a single thread that manages a queue of coroutines. When a coroutine hits an await, it suspends and yields control back — the loop runs the next ready coroutine. No threads, no locks.

import asyncio

async def say(msg, delay):
    await asyncio.sleep(delay)   # yields control here
    print(msg)

async def main():
    # Both run concurrently — total time ≈ 2s, not 3s
    await asyncio.gather(
        say("Hello", 1),
        say("World", 2),
    )

asyncio.run(main())

Coroutines, async/await Syntax

A coroutine is declared with async def. Calling it returns a coroutine object — it does NOT run until awaited.

async def fetch_data(url: str) -> dict:
    # This function is a coroutine
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

# WRONG — coroutine is created but never runs
result = fetch_data("https://api.example.com/data")

# CORRECT
result = await fetch_data("https://api.example.com/data")

Async Generators and Context Managers

async def paginate(base_url: str):
    page = 1
    while True:
        data = await fetch_data(f"{base_url}?page={page}")
        if not data:
            break
        for item in data:
            yield item
        page += 1

# Usage
async for record in paginate("https://api.example.com/users"):
    process(record)

Tasks and Gathering

Use asyncio.create_task() to schedule a coroutine without immediately awaiting it. Use asyncio.gather() to run many coroutines concurrently.

import asyncio
import aiohttp
import time

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

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in URLS]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    success = [r for r in results if not isinstance(r, Exception)]
    print(f"Fetched {len(success)} / {len(URLS)} successfully")

start = time.perf_counter()
asyncio.run(main())
print(f"Total time: {time.perf_counter() - start:.2f}s")
# Typical result: 50 HTTP requests in ~0.8s vs ~15s synchronous

Controlling Concurrency with Semaphores

async def main():
    sem = asyncio.Semaphore(10)  # max 10 concurrent requests

    async def fetch_limited(session, url):
        async with sem:
            return await fetch(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_limited(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

Real I/O: aiohttp, aiofiles, asyncpg

HTTP with aiohttp

pip install aiohttp

import aiohttp
import asyncio

async def post_data(session, url, payload):
    async with session.post(url, json=payload) as resp:
        resp.raise_for_status()
        return await resp.json()

async def main():
    async with aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=30),
        headers={"Authorization": "Bearer TOKEN"}
    ) as session:
        result = await post_data(session, "https://api.example.com/create", {"name": "test"})
        print(result)

File I/O with aiofiles

pip install aiofiles

import aiofiles
import asyncio

async def read_log(path: str) -> list[str]:
    async with aiofiles.open(path, "r") as f:
        content = await f.read()
    return content.splitlines()

async def write_results(path: str, lines: list[str]):
    async with aiofiles.open(path, "w") as f:
        await f.write("\n".join(lines))

PostgreSQL with asyncpg

pip install asyncpg

import asyncpg
import asyncio

async def main():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/mydb")
    
    # Parameterised query — SQL injection safe
    rows = await conn.fetch(
        "SELECT id, name, email FROM users WHERE active = $1 LIMIT $2",
        True, 100
    )
    
    for row in rows:
        print(dict(row))
    
    await conn.close()

asyncio.run(main())

Production Patterns

Connection Pool (asyncpg)

class Database:
    _pool: asyncpg.Pool = None

    @classmethod
    async def get_pool(cls) -> asyncpg.Pool:
        if cls._pool is None:
            cls._pool = await asyncpg.create_pool(
                dsn="postgresql://user:pass@localhost/mydb",
                min_size=5,
                max_size=20,
                command_timeout=60
            )
        return cls._pool

    @classmethod
    async def fetch(cls, query: str, *args):
        pool = await cls.get_pool()
        async with pool.acquire() as conn:
            return await conn.fetch(query, *args)

Async Queue for Worker Pool

import asyncio

async def worker(name: str, queue: asyncio.Queue):
    while True:
        item = await queue.get()
        try:
            await process(item)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=100)
    workers = [asyncio.create_task(worker(f"w{i}", queue)) for i in range(5)]
    
    for item in get_all_items():
        await queue.put(item)
    
    await queue.join()           # wait until all processed
    for w in workers:
        w.cancel()

Common Pitfalls

âš  Blocking calls inside async code — Never call time.sleep(), requests.get(), or any blocking function inside a coroutine. Use asyncio.sleep() and aiohttp instead, or offload to a thread pool with loop.run_in_executor().

# WRONG — blocks the entire event loop
async def bad():
    import requests
    return requests.get("https://example.com")

# CORRECT — offload blocking I/O to thread pool
async def good():
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, requests.get, "https://example.com")

✅ Use asyncio.run() at the top level. Never call loop.run_until_complete() in production code — it's the old pattern and doesn't handle cleanup properly.

Benchmark: Sync vs Async

import asyncio, aiohttp, requests, time

URLS = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 101)]

# Synchronous
def sync_fetch():
    return [requests.get(u).json() for u in URLS]

# Asynchronous
async def async_fetch():
    async with aiohttp.ClientSession() as s:
        tasks = [s.get(u) for u in URLS]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]

t = time.perf_counter(); sync_fetch()
print(f"Sync:  {time.perf_counter()-t:.2f}s")   # ~18.4s

t = time.perf_counter(); asyncio.run(async_fetch())
print(f"Async: {time.perf_counter()-t:.2f}s")   # ~0.9s

---

Key Takeaways: Use async/await for any I/O-bound Python code. Combine asyncio.gather() for fan-out concurrency, Semaphore for rate limiting, and connection pools for databases. Never block the event loop.

PythonAsyncasyncioaiohttpasyncpgConcurrencyBackend