Asynchronous Python: asyncio Fundamentals

Python’s Global Interpreter Lock (GIL) makes true parallelism difficult. But for I/O-bound work—HTTP requests, database queries, file operations—asyncio offers dramatic performance improvements.

Sync vs Async

Synchronous (Blocking)

import time
import requests

def fetch_url(url):
    response = requests.get(url)
    return response.text

def main():
    urls = ['http://example.com'] * 10
    start = time.time()
    
    for url in urls:
        fetch_url(url)  # Blocks until complete
    
    print(f"Time: {time.time() - start:.2f}s")

# ~10 seconds (1 second × 10 requests, sequential)

Asynchronous (Concurrent)

import asyncio
import aiohttp
import time

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

async def main():
    urls = ['http://example.com'] * 10
    start = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        await asyncio.gather(*tasks)  # Concurrent!
    
    print(f"Time: {time.time() - start:.2f}s")

asyncio.run(main())
# ~1 second (all requests overlap)

Core Concepts

Coroutines

Functions defined with async def:

async def my_coroutine():
    print("Start")
    await asyncio.sleep(1)  # Non-blocking sleep
    print("End")
    return "result"

# Calling returns a coroutine object (not the result!)
coro = my_coroutine()  # Nothing runs yet

# Must be awaited or run via event loop
result = asyncio.run(my_coroutine())

await

Suspends execution until the awaited coroutine completes:

async def main():
    result = await my_coroutine()  # Wait for completion
    print(result)

Event Loop

The engine that runs coroutines:

# Python 3.7+ (recommended)
asyncio.run(main())

# Manual control (advanced)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Concurrent Execution

asyncio.gather

Run coroutines concurrently, wait for all:

async def main():
    results = await asyncio.gather(
        fetch_url(url1),
        fetch_url(url2),
        fetch_url(url3),
    )
    return results  # [result1, result2, result3]

asyncio.create_task

Start a coroutine without immediately awaiting:

async def main():
    task1 = asyncio.create_task(fetch_url(url1))
    task2 = asyncio.create_task(fetch_url(url2))
    
    # Do other work while tasks run
    print("Tasks started")
    
    # Eventually await
    result1 = await task1
    result2 = await task2

asyncio.wait

More control over completion:

async def main():
    tasks = [asyncio.create_task(fetch_url(url)) for url in urls]
    
    # Wait for first to complete
    done, pending = await asyncio.wait(
        tasks, 
        return_when=asyncio.FIRST_COMPLETED
    )
    
    # Cancel remaining if needed
    for task in pending:
        task.cancel()

Timeouts and Cancellation

Timeouts

async def main():
    try:
        await asyncio.wait_for(slow_operation(), timeout=5.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

Cancellation

async def main():
    task = asyncio.create_task(long_running_operation())
    
    await asyncio.sleep(1)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

Async Context Managers and Iterators

Async Context Manager

class AsyncDatabase:
    async def __aenter__(self):
        await self.connect()
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        await self.disconnect()

async def main():
    async with AsyncDatabase() as db:
        await db.query("SELECT * FROM users")

Async Iterator

class AsyncRange:
    def __init__(self, n):
        self.n = n
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.n:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)  # Simulate async work
        self.current += 1
        return self.current

async def main():
    async for num in AsyncRange(5):
        print(num)

Common Libraries

HTTP: aiohttp

import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

Database: asyncpg (PostgreSQL)

import asyncpg

async def main():
    conn = await asyncpg.connect('postgresql://localhost/mydb')
    rows = await conn.fetch('SELECT * FROM users')
    await conn.close()

Files: aiofiles

import aiofiles

async def read_file(path):
    async with aiofiles.open(path, 'r') as f:
        return await f.read()

Mixing Sync and Async

Running Sync Code in Async

import asyncio
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    time.sleep(1)
    return "done"

async def main():
    loop = asyncio.get_event_loop()
    
    # Run in thread pool
    result = await loop.run_in_executor(
        ThreadPoolExecutor(),
        blocking_io
    )

Running Async from Sync

def sync_function():
    result = asyncio.run(async_function())
    return result

Debugging Async Code

Enable Debug Mode

asyncio.run(main(), debug=True)

Common Pitfalls

Forgetting await:

async def main():
    result = fetch_url(url)  # Returns coroutine, not result!
    result = await fetch_url(url)  # Correct

Blocking the event loop:

async def bad():
    time.sleep(1)  # Blocks everything!
    await asyncio.sleep(1)  # Correct

When to Use Async

Good use cases:

• HTTP clients/servers
• Database queries
• WebSocket connections
• File I/O
• Any I/O-bound work

Not ideal for:

• CPU-bound work (use multiprocessing)
• Simple scripts without I/O
• Libraries without async support

Final Thoughts

Asyncio adds complexity. The mental model is different from synchronous code. But for I/O-heavy applications, the performance gains are worth it.

Start with asyncio.gather for concurrent HTTP requests. Expand as needed. The async ecosystem is mature with libraries for most use cases.

Concurrency is about dealing with lots of things at once.