Async Programming & Python Ecosystem

Python's asyncio enables high-concurrency I/O without threads. Combined with its rich ecosystem — aiohttp, SQLAlchemy, Celery, and more — Python excels at building scalable web services and data pipelines.

asyncio Fundamentals

import asyncio

async def fetch(url: str, delay: float = 1.0) -> str:
    await asyncio.sleep(delay)   # non-blocking I/O simulation
    return f"Data from {url}"

async def main():
    # Sequential — total: 3 seconds
    r1 = await fetch("api1.com", 1.0)
    r2 = await fetch("api2.com", 2.0)

    # Concurrent with gather — total: 2 seconds (limited by slowest)
    results = await asyncio.gather(
        fetch("api1.com", 1.0),
        fetch("api2.com", 2.0),
        fetch("api3.com", 0.5),
    )
    print(results)

    # Ignore individual errors
    results = await asyncio.gather(
        fetch("api1.com"),
        fetch("broken.com"),
        return_exceptions=True,   # exceptions returned, not raised
    )

asyncio.run(main())

Real HTTP with aiohttp

import aiohttp
import asyncio

async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
    timeout = aiohttp.ClientTimeout(total=10)
    async with session.get(url, timeout=timeout) as resp:
        resp.raise_for_status()
        return await resp.json()

# Semaphore to cap concurrent requests
async def fetch_all(urls: list[str], limit: int = 10) -> list:
    semaphore = asyncio.Semaphore(limit)

    async def fetch_one(session, url):
        async with semaphore:
            return await fetch_json(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Async generator for pagination
async def paginate(base_url: str):
    page = 1
    async with aiohttp.ClientSession() as session:
        while True:
            data = await fetch_json(session, f"{base_url}?page={page}")
            if not data:
                break
            for item in data:
                yield item
            page += 1

Tasks & Patterns

import asyncio

# Tasks run concurrently in the background
async def main():
    task1 = asyncio.create_task(fetch("api1.com"))
    task2 = asyncio.create_task(fetch("api2.com"))

    # Other work runs while tasks execute
    await asyncio.sleep(0)

    await asyncio.wait_for(task1, timeout=5.0)
    results = await asyncio.gather(task1, task2)

# Process results as they arrive
async def process_first_ready(urls: list[str]) -> None:
    tasks = [asyncio.create_task(fetch(url)) for url in urls]
    for completed in asyncio.as_completed(tasks):
        result = await completed
        print(f"Got: {result}")

# Queue-based producer/consumer
async def producer(queue: asyncio.Queue, items: list) -> None:
    for item in items:
        await queue.put(item)
    await queue.put(None)   # sentinel

async def consumer(queue: asyncio.Queue) -> None:
    while (item := await queue.get()) is not None:
        print(f"Processing: {item}")
        queue.task_done()

Standard Library Essentials

# pathlib — modern file system
from pathlib import Path
project = Path('/myproject')
config = project / 'config' / 'settings.json'
config.parent.mkdir(parents=True, exist_ok=True)
text = config.read_text(encoding='utf-8')
for f in project.rglob('*.py'):
    print(f.stem)

# datetime with timezone
from datetime import datetime, timedelta, timezone
now = datetime.now(timezone.utc)
tomorrow = now + timedelta(days=1)
iso = now.isoformat()

# collections
from collections import Counter, defaultdict, deque
freq = Counter("the quick brown fox the fox".split())
print(freq.most_common(2))   # [('the', 2), ('fox', 2)]

graph = defaultdict(list)
graph['A'].append('B')        # no KeyError on missing key

history = deque(maxlen=5)    # circular buffer
for i in range(10):
    history.append(i)
print(list(history))          # [5, 6, 7, 8, 9]

Popular Libraries

• requests / httpx — HTTP clients; httpx adds async support and HTTP/2

• pydantic v2 — data validation and settings management with type annotations

• SQLAlchemy 2.0 — ORM and SQL toolkit, supports async with asyncpg

• celery + redis/rabbitmq — distributed task queues and background jobs

• pytest — testing framework with rich plugin ecosystem (pytest-asyncio, pytest-cov)

• ruff — ultra-fast linter and formatter (replaces flake8, black, isort)

• mypy / pyright — static type checkers

• click / typer — CLI frameworks (typer leverages type annotations)

• loguru — structured logging with minimal setup

Package Management

# Standard venv
python -m venv venv
source venv/bin/activate      # macOS/Linux
pip install fastapi uvicorn
pip freeze > requirements.txt

# uv — modern, Rust-based, very fast
pip install uv
uv venv && uv pip install fastapi uvicorn
uv pip sync requirements.txt  # exact reproducible installs

# Poetry — dependency management with lock files
poetry new myproject
poetry add fastapi sqlalchemy
poetry add --group dev pytest ruff mypy
poetry install && poetry run pytest