Add API Integration example (#6)

docs/plans/2026-03-08-api-integration-design.md

@@ -0,0 +1,48 @@
+# API Integration Example — Design
+
+**Date:** 2026-03-08
+**Pattern:** `trigger → api → transform → data`
+**Status:** Approved
+
+## Goal
+
+Sixth runnable example for workflow-patterns. A REST API data collector that fetches from APIs, transforms JSON responses, and stores structured results. Works offline with sample data fallback. Demonstrates `api → transform → data` — fetch, reshape, persist.
+
+## Architecture
+
+```
+select_integration() → fetch_api() → transform_response() → save_results()
+trigger                api            transform               data
+```
+
+## Module Structure
+
+```
+examples/api-integration/
+├── run.py                      # CLI entry point
+├── .env.example                # Optional API tokens
+├── .gitignore                  # .env, __pycache__, output/
+├── pyproject.toml              # Zero runtime dependencies (stdlib only)
+├── src/api_workflow/
+│   ├── __init__.py
+│   ├── models.py               # ApiConfig, ApiResponse, Integration
+│   ├── integrations.py         # 5 integration presets
+│   ├── client.py               # HTTP client with sample fallback (API layer)
+│   ├── transform.py            # JSON response transformers
+│   ├── storage.py              # Save results to file (data layer)
+│   └── display.py              # Terminal formatting
+├── tests/
+│   ├── test_models.py
+│   ├── test_integrations.py
+│   ├── test_client.py
+│   ├── test_transform.py
+│   ├── test_storage.py
+│   ├── test_display.py
+│   └── __init__.py
+└── output/                     # Saved API results
+```
+
+## Dependencies
+
+- Zero runtime dependencies (uses urllib.request from stdlib)
+- `pytest` (dev)

examples/api-integration/.env.example

@@ -0,0 +1,3 @@
+# API Integration — optional API tokens
+# Works without any keys (uses sample data fallback)
+# GITHUB_TOKEN=ghp_...

examples/api-integration/.gitignore

@@ -0,0 +1,4 @@
+.env
+__pycache__/
+output/
+*.pyc

examples/api-integration/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "api-integration"
+version = "0.1.0"
+description = "API Integration workflow: trigger -> api -> transform -> data"
+requires-python = ">=3.12"
+dependencies = []
+
+[dependency-groups]
+dev = ["pytest"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/api_workflow"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

examples/api-integration/run.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""API Integration workflow runner.
+
+Pattern: trigger -> api -> transform -> data
+
+Fetches data from REST APIs, transforms JSON responses,
+and saves structured results. Works offline with sample data.
+
+Usage:
+    uv run python run.py                      # interactive selection
+    uv run python run.py --integration 1      # select by number
+    uv run python run.py --format json        # save as JSON instead of CSV
+    uv run python run.py --live               # force live API call (no fallback)
+"""
+
+import argparse
+import os
+import sys
+from pathlib import Path
+
+from api_workflow.client import fetch
+from api_workflow.display import format_header, format_integration_menu, format_records_table
+from api_workflow.integrations import INTEGRATIONS, get_integration
+from api_workflow.storage import save_csv, save_json
+from api_workflow.transform import transform_response
+
+OUTPUT_DIR = Path(__file__).parent / "output"
+
+
+def _load_dotenv():
+    """Load .env file if it exists."""
+    env_path = Path(__file__).parent / ".env"
+    if not env_path.exists():
+        return
+    for line in env_path.read_text().splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        key, _, value = line.partition("=")
+        if key and value:
+            os.environ.setdefault(key.strip(), value.strip())
+
+
+def _select_integration() -> int:
+    """Interactive integration selection. Returns 0-based index."""
+    print(format_header("API Integration — Setup"))
+    print("Choose an integration:")
+    print(format_integration_menu(INTEGRATIONS))
+    print()
+    try:
+        choice = input(f"Integration (1-{len(INTEGRATIONS)}): ").strip()
+    except (EOFError, KeyboardInterrupt):
+        print("\nAborted.")
+        sys.exit(0)
+    try:
+        return int(choice) - 1
+    except ValueError:
+        return 0
+
+
+def main():
+    _load_dotenv()
+
+    parser = argparse.ArgumentParser(
+        description="API Integration: trigger -> api -> transform -> data"
+    )
+    parser.add_argument(
+        "--integration", type=int, default=None, help="Integration number (1-5)"
+    )
+    parser.add_argument(
+        "--format", choices=["csv", "json"], default="csv", help="Output format (default: csv)"
+    )
+    parser.add_argument(
+        "--live", action="store_true", help="Force live API call (no sample fallback)"
+    )
+    args = parser.parse_args()
+
+    # Step 1: Trigger — select integration
+    if args.integration is not None:
+        integration = get_integration(args.integration - 1)
+    else:
+        integration = get_integration(_select_integration())
+
+    print(f"\n── {integration.name} ──\n")
+
+    # Step 2: API — fetch data
+    sample = None if args.live else integration.sample_response
+    print(f"  Fetching from {integration.api.base_url}...")
+    response = fetch(integration.api, sample_data=sample)
+
+    if not response.is_success:
+        print(f"  Error: API request failed (status {response.status})")
+        sys.exit(1)
+
+    print(f"  Source: {response.source}")
+
+    # Step 3: Transform — reshape response
+    records = transform_response(response.data, integration.transform)
+    print(f"  Extracted {len(records)} records\n")
+
+    # Step 4: Data — save results
+    print(format_header("Results"))
+    print(format_records_table(records))
+
+    slug = integration.name.lower().replace(" ", "-")
+    if args.format == "json":
+        path = save_json(records, OUTPUT_DIR, slug)
+    else:
+        path = save_csv(records, OUTPUT_DIR, slug)
+
+    print(f"\n  Results saved to {path}\n")
+
+
+if __name__ == "__main__":
+    main()

examples/api-integration/src/api_workflow/client.py

@@ -0,0 +1,29 @@
+"""HTTP client for the API Integration workflow (API layer).
+
+Fetches from REST APIs with fallback to sample data.
+Uses urllib.request from stdlib — no external dependencies.
+"""
+
+import json
+import urllib.request
+import urllib.error
+
+from api_workflow.models import ApiConfig, ApiResponse
+
+
+def fetch_sample(sample_data: dict | list) -> ApiResponse:
+    """Return sample data as an ApiResponse."""
+    return ApiResponse(status=200, data=sample_data, source="sample")
+
+
+def fetch(cfg: ApiConfig, sample_data: dict | list | None = None) -> ApiResponse:
+    """Fetch from API, falling back to sample data on failure."""
+    try:
+        req = urllib.request.Request(cfg.full_url, headers=cfg.headers)
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            data = json.loads(resp.read().decode())
+            return ApiResponse(status=resp.status, data=data, source="live")
+    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError):
+        if sample_data is not None:
+            return fetch_sample(sample_data)
+        return ApiResponse(status=0, data={"error": "Request failed"}, source="error")

examples/api-integration/src/api_workflow/display.py

@@ -0,0 +1,52 @@
+"""Terminal display formatting for the API Integration workflow."""
+
+from api_workflow.models import Integration
+
+
+def format_header(title: str) -> str:
+    """Create a boxed ASCII header."""
+    width = max(len(title) + 8, 42)
+    top = "╔" + "═" * width + "╗"
+    mid = "║" + title.center(width) + "║"
+    bot = "╚" + "═" * width + "╝"
+    return f"\n{top}\n{mid}\n{bot}\n"
+
+
+def format_integration_menu(integrations: list[Integration]) -> str:
+    """Format integration selection menu."""
+    lines = []
+    for i, integ in enumerate(integrations, 1):
+        lines.append(f"  {i}. {integ.name:<22s} {integ.description}")
+    return "\n".join(lines)
+
+
+def format_records_table(records: list[dict], max_rows: int = 15) -> str:
+    """Format records as an ASCII table."""
+    if not records:
+        return "(no records)"
+
+    headers = list(records[0].keys())
+    widths = {h: len(h) for h in headers}
+    display = records[:max_rows]
+    for rec in display:
+        for h in headers:
+            widths[h] = max(widths[h], len(str(rec.get(h, ""))))
+
+    def row_str(values: dict) -> str:
+        cells = [str(values.get(h, "")).ljust(widths[h]) for h in headers]
+        return "│ " + " │ ".join(cells) + " │"
+
+    sep_top = "┌─" + "─┬─".join("─" * widths[h] for h in headers) + "─┐"
+    sep_mid = "├─" + "─┼─".join("─" * widths[h] for h in headers) + "─┤"
+    sep_bot = "└─" + "─┴─".join("─" * widths[h] for h in headers) + "─┘"
+
+    header_vals = {h: h for h in headers}
+    lines = [sep_top, row_str(header_vals), sep_mid]
+    for rec in display:
+        lines.append(row_str(rec))
+    lines.append(sep_bot)
+
+    if len(records) > max_rows:
+        lines.append(f"... and {len(records) - max_rows} more rows")
+
+    return "\n".join(lines)

examples/api-integration/src/api_workflow/integrations.py

@@ -0,0 +1,120 @@
+"""Integration presets for the API Integration workflow."""
+
+from api_workflow.models import ApiConfig, Integration, TransformSpec
+
+INTEGRATIONS: list[Integration] = [
+    Integration(
+        name="GitHub Repos",
+        description="Fetch popular repos, extract stars and language",
+        api=ApiConfig(
+            base_url="https://api.github.com",
+            endpoint="/search/repositories",
+            params={"q": "stars:>10000", "sort": "stars", "per_page": "10"},
+            headers={"Accept": "application/vnd.github.v3+json"},
+        ),
+        transform=TransformSpec(
+            extract_path="items",
+            fields=["full_name", "stargazers_count", "language", "description"],
+            rename={"stargazers_count": "stars", "full_name": "repo"},
+        ),
+        sample_response={
+            "items": [
+                {"full_name": "freeCodeCamp/freeCodeCamp", "stargazers_count": 385000, "language": "TypeScript", "description": "Open-source codebase and curriculum"},
+                {"full_name": "996icu/996.ICU", "stargazers_count": 269000, "language": "Rust", "description": "Repo for counting�"},
+                {"full_name": "EbookFoundation/free-programming-books", "stargazers_count": 315000, "language": None, "description": "Freely available programming books"},
+                {"full_name": "jwasham/coding-interview-university", "stargazers_count": 295000, "language": None, "description": "A complete computer science study plan"},
+                {"full_name": "sindresorhus/awesome", "stargazers_count": 290000, "language": None, "description": "Awesome lists about all kinds of topics"},
+            ]
+        },
+    ),
+    Integration(
+        name="Weather Forecast",
+        description="Fetch 7-day forecast for a city",
+        api=ApiConfig(
+            base_url="https://api.open-meteo.com",
+            endpoint="/v1/forecast",
+            params={"latitude": "52.52", "longitude": "13.41", "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum", "timezone": "Europe/Berlin"},
+        ),
+        transform=TransformSpec(
+            extract_path="daily",
+            fields=["time", "temperature_2m_max", "temperature_2m_min", "precipitation_sum"],
+            rename={"temperature_2m_max": "max_temp", "temperature_2m_min": "min_temp", "precipitation_sum": "rain_mm"},
+        ),
+        sample_response={
+            "daily": {
+                "time": ["2026-03-08", "2026-03-09", "2026-03-10", "2026-03-11", "2026-03-12", "2026-03-13", "2026-03-14"],
+                "temperature_2m_max": [8.2, 10.1, 12.5, 9.8, 7.3, 11.0, 13.2],
+                "temperature_2m_min": [2.1, 3.5, 5.0, 3.2, 1.8, 4.2, 6.1],
+                "precipitation_sum": [0.0, 2.3, 0.5, 5.1, 0.0, 0.0, 1.2],
+            }
+        },
+    ),
+    Integration(
+        name="Currency Rates",
+        description="Fetch exchange rates for major currencies",
+        api=ApiConfig(
+            base_url="https://open.er-api.com",
+            endpoint="/v6/latest/USD",
+        ),
+        transform=TransformSpec(
+            extract_path="rates",
+            fields=["EUR", "GBP", "JPY", "CHF", "CAD", "AUD"],
+        ),
+        sample_response={
+            "rates": {
+                "EUR": 0.92, "GBP": 0.79, "JPY": 149.50, "CHF": 0.88,
+                "CAD": 1.36, "AUD": 1.53, "CNY": 7.24, "INR": 83.12,
+            }
+        },
+    ),
+    Integration(
+        name="HackerNews Top",
+        description="Fetch top stories from Hacker News",
+        api=ApiConfig(
+            base_url="https://hacker-news.firebaseio.com",
+            endpoint="/v0/topstories.json",
+        ),
+        transform=TransformSpec(
+            extract_path="",
+            fields=[],
+        ),
+        sample_response={
+            "stories": [
+                {"title": "Show HN: A new approach to distributed systems", "score": 342, "by": "pg", "url": "https://example.com/1"},
+                {"title": "Why Rust is the future of systems programming", "score": 287, "by": "dang", "url": "https://example.com/2"},
+                {"title": "The hidden costs of microservices", "score": 256, "by": "tptacek", "url": "https://example.com/3"},
+                {"title": "Building a database from scratch in Go", "score": 198, "by": "jl", "url": "https://example.com/4"},
+                {"title": "A deep dive into WebAssembly performance", "score": 175, "by": "cw", "url": "https://example.com/5"},
+            ]
+        },
+    ),
+    Integration(
+        name="System Status",
+        description="Check service health and uptime",
+        api=ApiConfig(
+            base_url="https://status.example.com",
+            endpoint="/api/v1/services",
+        ),
+        transform=TransformSpec(
+            extract_path="services",
+            fields=["name", "status", "uptime", "response_time_ms"],
+            rename={"response_time_ms": "latency_ms"},
+        ),
+        sample_response={
+            "services": [
+                {"name": "API Gateway", "status": "operational", "uptime": "99.98%", "response_time_ms": 45},
+                {"name": "Database Primary", "status": "operational", "uptime": "99.99%", "response_time_ms": 12},
+                {"name": "Cache Layer", "status": "degraded", "uptime": "99.85%", "response_time_ms": 89},
+                {"name": "Auth Service", "status": "operational", "uptime": "99.97%", "response_time_ms": 23},
+                {"name": "CDN", "status": "operational", "uptime": "99.99%", "response_time_ms": 8},
+                {"name": "Search Index", "status": "maintenance", "uptime": "99.50%", "response_time_ms": 150},
+            ]
+        },
+    ),
+]
+
+
+def get_integration(index: int) -> Integration:
+    """Get integration by index, clamping to valid range."""
+    clamped = max(0, min(index, len(INTEGRATIONS) - 1))
+    return INTEGRATIONS[clamped]