Async I/O for Data Pipelines

Data pipelines often spend most of their time waiting — waiting for API responses, database queries, or file downloads. Async I/O lets you do useful work during that wait time, dramatically speeding up I/O-bound pipelines. This lesson covers asyncio fundamentals, aiohttp, async database drivers, concurrent API calls, and async generators.

---

Why Async?

SYNCHRONOUS                        ASYNCHRONOUS
┌─────┐  ┌─────┐  ┌─────┐        ┌─────┬─────┬─────┐
│ API │  │ API │  │ API │        │ API │ API │ API │
│  1  │  │  2  │  │  3  │        │  1  │  2  │  3  │
│ 2s  │  │ 2s  │  │ 2s  │        │ 2s  │ 2s  │ 2s  │
└─────┘  └─────┘  └─────┘        └─────┴─────┴─────┘
Total: 6 seconds                  Total: ~2 seconds

For I/O-bound work (API calls, database queries, file reads), async can provide near-linear speedup.

---

Asyncio Fundamentals

import asyncio

async def fetch_data(url: str) -> dict:
    """An async function (coroutine)."""
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # Simulates an I/O operation
    return {"url": url, "status": "ok"}

async def main():
    # Run a single coroutine
    result = await fetch_data("https://api.example.com/data")
    print(result)