HTTP & Consuming REST APIs¶

import httpx

# ── Basic GET with query params — always set a timeout ────────────
resp = httpx.get(
    "https://api.example.com/users",
    params={"active": "true", "limit": "50"},
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10.0,   # seconds; raises httpx.TimeoutException if exceeded
)
resp.raise_for_status()   # raises httpx.HTTPStatusError on 4xx/5xx
users = resp.json()       # parse JSON body → Python dict/list

# ── POST a JSON body to create a resource ────────────────────────
created = httpx.post(
    "https://api.example.com/orders",
    json={"product_id": "p42", "quantity": 3, "customer_id": "c99"},
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10.0,
)
print(created.status_code)   # 201 Created
print(created.headers["Location"])   # /orders/ord-8812

# ── PUT to fully replace a resource ──────────────────────────────
updated = httpx.put(
    "https://api.example.com/users/u1",
    json={"name": "Ada", "role": "admin", "active": True},
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10.0,
)
updated.raise_for_status()

# ── DELETE ────────────────────────────────────────────────────────
resp = httpx.delete(
    "https://api.example.com/users/u99",
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10.0,
)
assert resp.status_code == 204   # No Content — success, no body

import os
import httpx

# ── Pattern 1: API Key in header (e.g., Stripe, SendGrid) ─────────
API_KEY = os.environ["EXTERNAL_API_KEY"]   # never hard-code credentials

resp = httpx.get(
    "https://api.example.com/reports",
    headers={"X-API-Key": API_KEY},
    timeout=10.0,
)

# ── Pattern 2: Bearer token (OAuth2 / JWT) ────────────────────────
ACCESS_TOKEN = os.environ["ACCESS_TOKEN"]

resp = httpx.get(
    "https://api.example.com/data",
    headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
    timeout=10.0,
)

# ── Pattern 3: OAuth2 Client Credentials (service-to-service) ─────
def get_access_token(client_id: str, client_secret: str, token_url: str) -> str:
    """Exchange client credentials for a short-lived access token."""
    resp = httpx.post(
        token_url,
        data={
            "grant_type":    "client_credentials",
            "client_id":     client_id,
            "client_secret": client_secret,
            "scope":         "read:events write:orders",
        },
        timeout=10.0,
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

# Use short-lived token; refresh before it expires (check "expires_in" field)
token = get_access_token(
    os.environ["CLIENT_ID"],
    os.environ["CLIENT_SECRET"],
    "https://auth.example.com/oauth/token",
)
resp = httpx.get(
    "https://api.example.com/protected",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
)

import time, uuid, httpx
from typing import Any

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def post_with_retry(
    url: str,
    payload: dict[str, Any],
    token: str,
    max_attempts: int = 4,
    base_delay: float = 0.5,
) -> dict[str, Any]:
    """
    POST with:
    - Idempotency key (safe retries for non-idempotent endpoint)
    - Exponential backoff on retryable errors
    - Timeout on every request
    """
    idempotency_key = str(uuid.uuid4())   # unique per logical operation
    last_exc: Exception | None = None

    for attempt in range(max_attempts):
        try:
            resp = httpx.post(
                url,
                json=payload,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Idempotency-Key": idempotency_key,
                },
                timeout=10.0,
            )
            if resp.status_code in RETRYABLE_STATUS:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
                continue
            resp.raise_for_status()
            return resp.json()
        except httpx.TimeoutException as e:
            last_exc = e
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)

    raise RuntimeError(f"All {max_attempts} attempts failed") from last_exc