↓ Skip to main content
Meshack Mogire

Mastering Python Async: Beyond Basic Coroutines

Python’s asyncio is powerful, but often misunderstood. Let’s move beyond simple await calls and understand the Event Loop.

1. The Event Loop Explained
#

Think of the event loop as a single-threaded manager. It can only do one thing at a time, but it’s very fast at switching context when waiting for I/O.

Key Concept: Async code doesn’t make CPU-bound tasks faster. It makes I/O-bound tasks (network requests, DB queries) non-blocking.

Basic Pattern
#

import asyncio

async def fetch_data():
    print("Fetching...")
    await asyncio.sleep(2)  # Simulates I/O
    print("Done!")
    return {"data": 123}

async def main():
    # Run concurrently - This takes 2 seconds total, not 6
    results = await asyncio.gather(fetch_data(), fetch_data(), fetch_data())
    print(results)

if __name__ == "__main__":
    asyncio.run(main())

2. Common Pitfalls
#

Pitfall #1: Blocking the Loop
#

Running heavy calculations (like image processing) inside an async function freezes the entire app.

Bad:

async def process_image():
    # This blocks everything!
    time.sleep(5)

Good:

async def process_image():
    loop = asyncio.get_running_loop()
    # Run in a separate thread
    await loop.run_in_executor(None, do_heavy_work)

Pitfall #2: Context Variables
#

threading.local() doesn’t work in async. Use contextvars.

import contextvars

request_id = contextvars.ContextVar("request_id")

async def log(message):
    print(f"[{request_id.get()}] {message}")

3. Profiling Async Code
#

How do you know what’s slow? cProfile often fails with async.

Use py-spy or built-in debug mode.

PYTHONASYNCIODEBUG=1 python my_script.py

This will warn you if a coroutine blocks the loop for too long.

Executing <Task...> took 0.150 seconds

4. Real World Async: Aiohttp
#

Calling 100 APIs sequentially takes 100 seconds. Concurrently, it takes 1 second.

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, f"https://api.com/items/{i}") for i in range(100)]
        # Run all 100 requests at once
        responses = await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

When to Use Async?
#

  • Web Scrapers: Fetching 1000 URLs? Async is 10x faster than sync.
  • Chat Servers: Handling 10k connections? Async is mandatory (WebSockets).
  • Microservices: Calling 5 downstream APIs? asyncio.gather is your best friend.