Working with JSON APIs

import json
from pathlib import Path

# Parse a JSON string
data = json.loads('{"name": "web-01", "status": "healthy", "uptime": 99.9}')
print(data["name"])  # web-01

# Parse a JSON file
config = json.loads(Path("config.json").read_text())

# Serialize to JSON string
output = json.dumps(data, indent=2, sort_keys=True)

# Write to file
Path("output.json").write_text(json.dumps(data, indent=2) + "\n")

import requests

def parse_api_response(resp: requests.Response) -> dict | list:
    """Parse a JSON API response with error handling."""
    # Check content type
    content_type = resp.headers.get("Content-Type", "")
    if "application/json" not in content_type:
        raise ValueError(f"Expected JSON, got {content_type}")

    # Parse JSON
    try:
        return resp.json()
    except requests.exceptions.JSONDecodeError:
        raise ValueError(f"Invalid JSON: {resp.text[:200]}")

# Usage
resp = requests.get("https://api.github.com/user", headers={
    "Accept": "application/vnd.github.v3+json",
})

if resp.ok:
    user = parse_api_response(resp)
    print(f"Logged in as: {user['login']}")
else:
    error = parse_api_response(resp)
    print(f"API error: {error.get('message', 'Unknown')}")

def get_nested(data: dict, *keys, default=None):
    """Safely access nested dictionary keys."""
    current = data
    for key in keys:
        if isinstance(current, dict):
            current = current.get(key, default)
        else:
            return default
    return current

# API response with deeply nested data
response = {
    "data": {
        "items": [
            {
                "metadata": {
                    "name": "web-01",
                    "labels": {"env": "production", "team": "platform"},
                },
                "status": {"phase": "Running"},
            }
        ]
    }
}

# Safe access — no KeyError, no TypeError
name = get_nested(response, "data", "items", 0, "metadata", "name")
env = get_nested(response, "data", "items", 0, "metadata", "labels", "env")
phase = get_nested(response, "data", "items", 0, "status", "phase")

print(f"{name} ({env}): {phase}")

import requests

class APIError(Exception):
    def __init__(self, status_code: int, message: str, details: dict | None = None):
        self.status_code = status_code
        self.message = message
        self.details = details or {}
        super().__init__(f"HTTP {status_code}: {message}")

def request_or_raise(method: str, url: str, **kwargs) -> dict:
    """Make an API request and raise on errors."""
    resp = requests.request(method, url, timeout=10, **kwargs)

    if resp.ok:
        return resp.json() if resp.content else {}

    try:
        error_data = resp.json()
        message = error_data.get("message", error_data.get("error", str(error_data)))
    except Exception:
        message = resp.text[:200]

    raise APIError(resp.status_code, message, error_data if resp.content else {})

# Usage
try:
    repo = request_or_raise("GET", "https://api.github.com/repos/org/nonexistent")
except APIError as e:
    if e.status_code == 404:
        print("Repository not found")
    elif e.status_code == 403:
        print("Rate limited or insufficient permissions")
    else:
        print(f"API error: {e.message}")

import requests

# Offset-based pagination
def fetch_offset_pages(url: str, token: str) -> list[dict]:
    items = []
    offset = 0
    limit = 100
    while True:
        resp = requests.get(url, params={"offset": offset, "limit": limit},
                          headers={"Authorization": f"Bearer {token}"})
        resp.raise_for_status()
        data = resp.json()
        items.extend(data["results"])
        if not data.get("has_next"):
            break
        offset += limit
    return items

# Cursor-based pagination (preferred for large datasets)
def fetch_cursor_pages(url: str, token: str) -> list[dict]:
    items = []
    cursor = None
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor
        resp = requests.get(url, params=params,
                          headers={"Authorization": f"Bearer {token}"})
        resp.raise_for_status()
        data = resp.json()
        items.extend(data["items"])
        cursor = data.get("next_cursor")
        if not cursor:
            break
    return items

from pydantic import BaseModel, Field
import requests

class ServerHealth(BaseModel):
    name: str
    status: str
    cpu_percent: float = Field(ge=0, le=100)
    memory_percent: float = Field(ge=0, le=100)
    uptime_seconds: int

class HealthResponse(BaseModel):
    servers: list[ServerHealth]
    timestamp: str
    overall_status: str

def get_cluster_health(api_url: str) -> HealthResponse:
    resp = requests.get(f"{api_url}/health", timeout=10)
    resp.raise_for_status()
    # Validates and parses in one step
    return HealthResponse(**resp.json())

# Invalid data raises ValidationError with clear messages
health = get_cluster_health("https://api.example.com")
for server in health.servers:
    if server.status != "healthy":
        print(f"Unhealthy: {server.name}")