requests & httpx

import requests

# GET request
resp = requests.get("https://api.example.com/health")
print(resp.status_code)  # 200
print(resp.json())       # Parse JSON response
print(resp.text)         # Raw text

# POST with JSON body
resp = requests.post(
    "https://api.example.com/deploy",
    json={"service": "web-01", "replicas": 3},
    headers={"Authorization": "Bearer token123"},
)

# PUT and DELETE
requests.put("https://api.example.com/config", json={"key": "value"})
requests.delete("https://api.example.com/servers/web-02")

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session() -> requests.Session:
    """Create a session with automatic retries."""
    session = requests.Session()

    retry = Retry(
        total=3,
        backoff_factor=0.5,  # 0.5s, 1s, 2s
        status_forcelist=[429, 500, 502, 503, 504],
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    return session

# Reuse the session across requests (connection pooling)
session = create_session()
for server in ["web-01", "web-02", "web-03"]:
    resp = session.get(f"http://{server}:8080/health")
    print(f"{server}: {resp.status_code}")

import requests
from requests.auth import HTTPBasicAuth

# Bearer token (most common for APIs)
resp = requests.get(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9..."},
)

# Basic auth
resp = requests.get(
    "https://api.example.com/admin",
    auth=HTTPBasicAuth("admin", "password"),
)

# API key in header
resp = requests.get(
    "https://api.example.com/data",
    headers={"X-API-Key": "my-api-key-123"},
)

# API key as query parameter
resp = requests.get(
    "https://api.example.com/data",
    params={"api_key": "my-api-key-123"},
)

# Token from environment variable
import os
token = os.environ["API_TOKEN"]
resp = requests.get(
    "https://api.example.com/data",
    headers={"Authorization": f"Bearer {token}"},
)

import requests

def api_request(method: str, url: str, **kwargs) -> dict:
    """Make an API request with proper error handling."""
    try:
        resp = requests.request(method, url, timeout=10, **kwargs)
        resp.raise_for_status()  # Raises HTTPError for 4xx/5xx
        return resp.json()
    except requests.exceptions.Timeout:
        print(f"Request timed out: {url}")
        raise
    except requests.exceptions.ConnectionError:
        print(f"Connection failed: {url}")
        raise
    except requests.exceptions.HTTPError as e:
        print(f"HTTP {resp.status_code}: {resp.text}")
        raise
    except requests.exceptions.JSONDecodeError:
        print(f"Invalid JSON response from {url}")
        raise

# Usage
try:
    data = api_request("GET", "https://api.example.com/health")
except requests.exceptions.RequestException as e:
    print(f"API request failed: {e}")

import httpx
import asyncio

# Sync usage (drop-in replacement for requests)
with httpx.Client() as client:
    resp = client.get("https://api.example.com/health")
    print(resp.json())

# Async usage
async def check_services(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        results = []
        for url, resp in zip(urls, responses):
            if isinstance(resp, Exception):
                results.append({"url": url, "error": str(resp)})
            else:
                results.append({"url": url, "status": resp.status_code})
        return results

# Run async function
results = asyncio.run(check_services([
    "http://web-01:8080/health",
    "http://web-02:8080/health",
]))

import requests
import time

def paginated_get(url: str, token: str, page_size: int = 100) -> list[dict]:
    """Fetch all pages from a paginated API."""
    all_items = []
    headers = {"Authorization": f"Bearer {token}"}
    params = {"per_page": page_size, "page": 1}

    while True:
        resp = requests.get(url, headers=headers, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()

        items = data.get("items", [])
        all_items.extend(items)

        if len(items) < page_size:
            break
        params["page"] += 1

    return all_items

def rate_limited_request(session: requests.Session, url: str, max_retries: int = 3) -> dict:
    """Handle rate limiting with exponential backoff."""
    for attempt in range(max_retries):
        resp = session.get(url, timeout=10)
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
            time.sleep(retry_after)
            continue
        resp.raise_for_status()
        return resp.json()
    raise Exception(f"Rate limited after {max_retries} retries")