{"slug":"python-async-and-cli-patterns","title":"Python Async Patterns, Click CLIs, and Production Script Conventions","tags":["python","async","asyncio","click","cli","httpx","scripts"],"agent_summary":"Python production patterns — asyncio concurrency, httpx async HTTP, Click CLI framework, set -euo pipefail Bash equivalents, error handling, rate limiting, progress bars, and script conventions for automation pipelines.","trigger_phrases":["Python async","asyncio","httpx async","Click CLI","Python CLI","Python script","async Python","Python rate limit","Python pipeline"],"runnable":false,"markdown":"\n## Overview\n\nProduction Python for automation, data pipelines, and API clients. Key stack: asyncio for concurrency, httpx for async HTTP, Click for CLIs, tenacity for retry logic.\n\n## Async Foundations\n\n```python\nimport asyncio\nimport httpx\n\nasync def fetch(client: httpx.AsyncClient, url: str) -> dict:\n r = await client.get(url)\n r.raise_for_status()\n return r.json()\n\nasync def main():\n async with httpx.AsyncClient(timeout=30) as client:\n # Sequential\n data = await fetch(client, \"https://api.example.com/users\")\n\n # Concurrent (all at once)\n results = await asyncio.gather(\n fetch(client, \"https://api.example.com/users\"),\n fetch(client, \"https://api.example.com/projects\"),\n fetch(client, \"https://api.example.com/settings\"),\n )\n\nasyncio.run(main())\n```\n\n## Bounded Concurrency\n\nAvoid overwhelming APIs. Use a semaphore to cap parallel requests:\n\n```python\nasync def fetch_all(urls: list[str], max_concurrent: int = 5) -> list[dict]:\n semaphore = asyncio.Semaphore(max_concurrent)\n\n async def bounded_fetch(client, url):\n async with semaphore:\n return await fetch(client, url)\n\n async with httpx.AsyncClient() as client:\n return await asyncio.gather(\n *[bounded_fetch(client, url) for url in urls]\n )\n```\n\n## Rate Limiting\n\n```python\nimport asyncio\nfrom collections import deque\nimport time\n\nclass RateLimiter:\n def __init__(self, calls_per_second: float):\n self.min_interval = 1.0 / calls_per_second\n self.last_call = 0.0\n\n async def wait(self):\n now = time.monotonic()\n elapsed = now - self.last_call\n if elapsed < self.min_interval:\n await asyncio.sleep(self.min_interval - elapsed)\n self.last_call = time.monotonic()\n\nlimiter = RateLimiter(calls_per_second=2)\n\nasync def rate_limited_fetch(client, url):\n await limiter.wait()\n return await fetch(client, url)\n```\n\n## Retry Logic (tenacity)\n\n```python\nfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type\n\n@retry(\n stop=stop_after_attempt(3),\n wait=wait_exponential(multiplier=1, min=1, max=30),\n retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)),\n)\nasync def fetch_with_retry(client: httpx.AsyncClient, url: str) -> dict:\n r = await client.get(url)\n if r.status_code == 429:\n retry_after = int(r.headers.get(\"Retry-After\", 60))\n await asyncio.sleep(retry_after)\n r.raise_for_status()\n r.raise_for_status()\n return r.json()\n```\n\n## Click CLI Framework\n\n```python\nimport click\nimport asyncio\n\n@click.group()\n@click.option(\"--verbose\", \"-v\", is_flag=True, help=\"Enable verbose output\")\n@click.pass_context\ndef cli(ctx, verbose):\n ctx.ensure_object(dict)\n ctx.obj[\"verbose\"] = verbose\n\n@cli.command()\n@click.argument(\"url\")\n@click.option(\"--output\", \"-o\", default=\"result.json\", help=\"Output file path\")\n@click.option(\"--limit\", \"-n\", default=100, show_default=True, help=\"Max results\")\n@click.pass_context\ndef fetch(ctx, url, output, limit):\n \"\"\"Fetch data from URL and save to file.\"\"\"\n verbose = ctx.obj[\"verbose\"]\n if verbose:\n click.echo(f\"Fetching {url} (limit={limit})\")\n\n # run async from sync Click handler\n result = asyncio.run(_fetch_data(url, limit))\n\n import json\n with open(output, \"w\") as f:\n json.dump(result, f, indent=2)\n click.echo(f\"Saved {len(result)} items to {output}\")\n\nasync def _fetch_data(url: str, limit: int) -> list:\n async with httpx.AsyncClient() as client:\n r = await client.get(url, params={\"limit\": limit})\n r.raise_for_status()\n return r.json()\n\nif __name__ == \"__main__\":\n cli()\n```\n\n## Progress Bars (tqdm)\n\n```python\nfrom tqdm.asyncio import tqdm_asyncio\n\nasync def process_all(items: list) -> list:\n results = []\n async for result in tqdm_asyncio.as_completed(\n [process(item) for item in items],\n total=len(items),\n desc=\"Processing\",\n ):\n results.append(result)\n return results\n```\n\n## Error Handling Conventions\n\n```python\nimport logging\nimport sys\n\nlogging.basicConfig(\n level=logging.INFO,\n format=\"%(asctime)s %(levelname)s %(name)s — %(message)s\",\n)\nlog = logging.getLogger(__name__)\n\nasync def safe_run():\n try:\n await main()\n except KeyboardInterrupt:\n log.info(\"Interrupted by user\")\n sys.exit(0)\n except httpx.HTTPStatusError as e:\n log.error(f\"HTTP {e.response.status_code}: {e.request.url}\")\n sys.exit(1)\n except Exception as e:\n log.exception(f\"Unexpected error: {e}\")\n sys.exit(1)\n```\n\n## Script Entry Point Convention\n\n```python\n# Always end scripts with\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\nFor Click apps:\n```python\nif __name__ == \"__main__\":\n cli()\n```\n\n## uv for Script Dependencies\n\n```bash\n# Run script with inline dependencies (no virtualenv needed)\nuv run --with httpx --with click python my-script.py\n\n# Or with a pyproject.toml inline\n# /// script\n# dependencies = [\"httpx\", \"click\", \"tqdm\"]\n# ///\n```\n\n## Pagination Helper\n\n```python\nasync def paginate(client: httpx.AsyncClient, url: str, page_size: int = 100):\n \"\"\"Yield all pages from a paginated API.\"\"\"\n page = 1\n while True:\n r = await client.get(url, params={\"page\": page, \"limit\": page_size})\n r.raise_for_status()\n data = r.json()\n items = data.get(\"items\", data) # handle both formats\n if not items:\n break\n yield items\n if len(items) < page_size:\n break\n page += 1\n\n# Usage\nasync def fetch_all_users():\n results = []\n async with httpx.AsyncClient() as client:\n async for page in paginate(client, \"https://api.example.com/users\"):\n results.extend(page)\n return results\n```\n\n## Structured Output Pattern\n\n```python\nimport json\nfrom dataclasses import dataclass, asdict\nfrom typing import Optional\n\n@dataclass\nclass Result:\n id: str\n name: str\n status: str\n error: Optional[str] = None\n\nresults: list[Result] = []\n# ... populate results\n\n# Write JSONL for large datasets\nwith open(\"output.jsonl\", \"w\") as f:\n for r in results:\n f.write(json.dumps(asdict(r)) + \"\\n\")\n```\n","html":"

Overview

\n

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.

\n

Async Foundations

\n
import asyncio\nimport httpx\n\nasync def fetch(client: httpx.AsyncClient, url: str) -> dict:\n    r = await client.get(url)\n    r.raise_for_status()\n    return r.json()\n\nasync def main():\n    async with httpx.AsyncClient(timeout=30) as client:\n        # Sequential\n        data = await fetch(client, \"https://api.example.com/users\")\n\n        # Concurrent (all at once)\n        results = await asyncio.gather(\n            fetch(client, \"https://api.example.com/users\"),\n            fetch(client, \"https://api.example.com/projects\"),\n            fetch(client, \"https://api.example.com/settings\"),\n        )\n\nasyncio.run(main())\n
\n

Bounded Concurrency

\n

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

\n
async def fetch_all(urls: list[str], max_concurrent: int = 5) -> list[dict]:\n    semaphore = asyncio.Semaphore(max_concurrent)\n\n    async def bounded_fetch(client, url):\n        async with semaphore:\n            return await fetch(client, url)\n\n    async with httpx.AsyncClient() as client:\n        return await asyncio.gather(\n            *[bounded_fetch(client, url) for url in urls]\n        )\n
\n

Rate Limiting

\n
import asyncio\nfrom collections import deque\nimport time\n\nclass RateLimiter:\n    def __init__(self, calls_per_second: float):\n        self.min_interval = 1.0 / calls_per_second\n        self.last_call = 0.0\n\n    async def wait(self):\n        now = time.monotonic()\n        elapsed = now - self.last_call\n        if elapsed < self.min_interval:\n            await asyncio.sleep(self.min_interval - elapsed)\n        self.last_call = time.monotonic()\n\nlimiter = RateLimiter(calls_per_second=2)\n\nasync def rate_limited_fetch(client, url):\n    await limiter.wait()\n    return await fetch(client, url)\n
\n

Retry Logic (tenacity)

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

Click CLI Framework

\n
import click\nimport asyncio\n\n@click.group()\n@click.option(\"--verbose\", \"-v\", is_flag=True, help=\"Enable verbose output\")\n@click.pass_context\ndef cli(ctx, verbose):\n    ctx.ensure_object(dict)\n    ctx.obj[\"verbose\"] = verbose\n\n@cli.command()\n@click.argument(\"url\")\n@click.option(\"--output\", \"-o\", default=\"result.json\", help=\"Output file path\")\n@click.option(\"--limit\", \"-n\", default=100, show_default=True, help=\"Max results\")\n@click.pass_context\ndef fetch(ctx, url, output, limit):\n    \"\"\"Fetch data from URL and save to file.\"\"\"\n    verbose = ctx.obj[\"verbose\"]\n    if verbose:\n        click.echo(f\"Fetching {url} (limit={limit})\")\n\n    # run async from sync Click handler\n    result = asyncio.run(_fetch_data(url, limit))\n\n    import json\n    with open(output, \"w\") as f:\n        json.dump(result, f, indent=2)\n    click.echo(f\"Saved {len(result)} items to {output}\")\n\nasync def _fetch_data(url: str, limit: int) -> list:\n    async with httpx.AsyncClient() as client:\n        r = await client.get(url, params={\"limit\": limit})\n        r.raise_for_status()\n        return r.json()\n\nif __name__ == \"__main__\":\n    cli()\n
\n

Progress Bars (tqdm)

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

Error Handling Conventions

\n
import logging\nimport sys\n\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s %(levelname)s %(name)s — %(message)s\",\n)\nlog = logging.getLogger(__name__)\n\nasync def safe_run():\n    try:\n        await main()\n    except KeyboardInterrupt:\n        log.info(\"Interrupted by user\")\n        sys.exit(0)\n    except httpx.HTTPStatusError as e:\n        log.error(f\"HTTP {e.response.status_code}: {e.request.url}\")\n        sys.exit(1)\n    except Exception as e:\n        log.exception(f\"Unexpected error: {e}\")\n        sys.exit(1)\n
\n

Script Entry Point Convention

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

For Click apps:

\n
if __name__ == \"__main__\":\n    cli()\n
\n

uv for Script Dependencies

\n
# Run script with inline dependencies (no virtualenv needed)\nuv run --with httpx --with click python my-script.py\n\n# Or with a pyproject.toml inline\n# /// script\n# dependencies = [\"httpx\", \"click\", \"tqdm\"]\n# ///\n
\n

Pagination Helper

\n
async def paginate(client: httpx.AsyncClient, url: str, page_size: int = 100):\n    \"\"\"Yield all pages from a paginated API.\"\"\"\n    page = 1\n    while True:\n        r = await client.get(url, params={\"page\": page, \"limit\": page_size})\n        r.raise_for_status()\n        data = r.json()\n        items = data.get(\"items\", data)  # handle both formats\n        if not items:\n            break\n        yield items\n        if len(items) < page_size:\n            break\n        page += 1\n\n# Usage\nasync def fetch_all_users():\n    results = []\n    async with httpx.AsyncClient() as client:\n        async for page in paginate(client, \"https://api.example.com/users\"):\n            results.extend(page)\n    return results\n
\n

Structured Output Pattern

\n
import json\nfrom dataclasses import dataclass, asdict\nfrom typing import Optional\n\n@dataclass\nclass Result:\n    id: str\n    name: str\n    status: str\n    error: Optional[str] = None\n\nresults: list[Result] = []\n# ... populate results\n\n# Write JSONL for large datasets\nwith open(\"output.jsonl\", \"w\") as f:\n    for r in results:\n        f.write(json.dumps(asdict(r)) + \"\\n\")\n
\n"}