Python Async Patterns, Click CLIs, and Production Script Conventions

Overview

Production Python for automation, data pipelines, and API clients. Key stack: asyncio for concurrency, httpx for async HTTP, Click for CLIs, tenacity for retry logic.

Async Foundations

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url)
    r.raise_for_status()
    return r.json()

async def main():
    async with httpx.AsyncClient(timeout=30) as client:
        # Sequential
        data = await fetch(client, "https://api.example.com/users")

        # Concurrent (all at once)
        results = await asyncio.gather(
            fetch(client, "https://api.example.com/users"),
            fetch(client, "https://api.example.com/projects"),
            fetch(client, "https://api.example.com/settings"),
        )

asyncio.run(main())

Bounded Concurrency

Avoid overwhelming APIs. Use a semaphore to cap parallel requests:

async def fetch_all(urls: list[str], max_concurrent: int = 5) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(client, url):
        async with semaphore:
            return await fetch(client, url)

    async with httpx.AsyncClient() as client:
        return await asyncio.gather(
            *[bounded_fetch(client, url) for url in urls]
        )

Rate Limiting

import asyncio
from collections import deque
import time

class RateLimiter:
    def __init__(self, calls_per_second: float):
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0.0

    async def wait(self):
        now = time.monotonic()
        elapsed = now - self.last_call
        if elapsed < self.min_interval:
            await asyncio.sleep(self.min_interval - elapsed)
        self.last_call = time.monotonic()

limiter = RateLimiter(calls_per_second=2)

async def rate_limited_fetch(client, url):
    await limiter.wait()
    return await fetch(client, url)

Retry Logic (tenacity)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)),
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url)
    if r.status_code == 429:
        retry_after = int(r.headers.get("Retry-After", 60))
        await asyncio.sleep(retry_after)
        r.raise_for_status()
    r.raise_for_status()
    return r.json()

Click CLI Framework

import click
import asyncio

@click.group()
@click.option("--verbose", "-v", is_flag=True, help="Enable verbose output")
@click.pass_context
def cli(ctx, verbose):
    ctx.ensure_object(dict)
    ctx.obj["verbose"] = verbose

@cli.command()
@click.argument("url")
@click.option("--output", "-o", default="result.json", help="Output file path")
@click.option("--limit", "-n", default=100, show_default=True, help="Max results")
@click.pass_context
def fetch(ctx, url, output, limit):
    """Fetch data from URL and save to file."""
    verbose = ctx.obj["verbose"]
    if verbose:
        click.echo(f"Fetching {url} (limit={limit})")

    # run async from sync Click handler
    result = asyncio.run(_fetch_data(url, limit))

    import json
    with open(output, "w") as f:
        json.dump(result, f, indent=2)
    click.echo(f"Saved {len(result)} items to {output}")

async def _fetch_data(url: str, limit: int) -> list:
    async with httpx.AsyncClient() as client:
        r = await client.get(url, params={"limit": limit})
        r.raise_for_status()
        return r.json()

if __name__ == "__main__":
    cli()

Progress Bars (tqdm)

from tqdm.asyncio import tqdm_asyncio

async def process_all(items: list) -> list:
    results = []
    async for result in tqdm_asyncio.as_completed(
        [process(item) for item in items],
        total=len(items),
        desc="Processing",
    ):
        results.append(result)
    return results

Error Handling Conventions

import logging
import sys

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s — %(message)s",
)
log = logging.getLogger(__name__)

async def safe_run():
    try:
        await main()
    except KeyboardInterrupt:
        log.info("Interrupted by user")
        sys.exit(0)
    except httpx.HTTPStatusError as e:
        log.error(f"HTTP {e.response.status_code}: {e.request.url}")
        sys.exit(1)
    except Exception as e:
        log.exception(f"Unexpected error: {e}")
        sys.exit(1)

Script Entry Point Convention

# Always end scripts with
if __name__ == "__main__":
    asyncio.run(main())

For Click apps:

if __name__ == "__main__":
    cli()

uv for Script Dependencies

# Run script with inline dependencies (no virtualenv needed)
uv run --with httpx --with click python my-script.py

# Or with a pyproject.toml inline
# /// script
# dependencies = ["httpx", "click", "tqdm"]
# ///

Pagination Helper

async def paginate(client: httpx.AsyncClient, url: str, page_size: int = 100):
    """Yield all pages from a paginated API."""
    page = 1
    while True:
        r = await client.get(url, params={"page": page, "limit": page_size})
        r.raise_for_status()
        data = r.json()
        items = data.get("items", data)  # handle both formats
        if not items:
            break
        yield items
        if len(items) < page_size:
            break
        page += 1

# Usage
async def fetch_all_users():
    results = []
    async with httpx.AsyncClient() as client:
        async for page in paginate(client, "https://api.example.com/users"):
            results.extend(page)
    return results

Structured Output Pattern

import json
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class Result:
    id: str
    name: str
    status: str
    error: Optional[str] = None

results: list[Result] = []
# ... populate results

# Write JSONL for large datasets
with open("output.jsonl", "w") as f:
    for r in results:
        f.write(json.dumps(asdict(r)) + "\n")