HTTP & APIs

uv add httpx

import httpx

# GET
response = httpx.get("https://api.example.com/endpoints")
print(response.status_code)  # 200
print(response.json())       # Parsed JSON

# POST with JSON body
response = httpx.post(
    "https://api.example.com/endpoints",
    json={"mac": "00:11:22:33:44:55", "group": "Employees"}
)

# PUT
response = httpx.put(
    "https://api.example.com/endpoints/123",
    json={"group": "Guests"}
)

# DELETE
response = httpx.delete("https://api.example.com/endpoints/123")

import httpx

response = httpx.get("https://api.example.com/data")

# Status
response.status_code      # 200
response.is_success       # True (2xx)
response.is_error         # False
response.is_redirect      # False

# Headers
response.headers          # Headers dict
response.headers["content-type"]

# Content
response.text             # String content
response.content          # Bytes
response.json()           # Parsed JSON

# Raise on error
response.raise_for_status()  # Raises HTTPStatusError if 4xx/5xx

import httpx

# Context manager (recommended)
with httpx.Client() as client:
    r1 = client.get("https://api.example.com/users")
    r2 = client.get("https://api.example.com/endpoints")
# Connection closed automatically

# Reusable client
client = httpx.Client(
    base_url="https://ise-01:9060",
    timeout=30.0,
    verify=False  # Disable SSL verification (dev only)
)

try:
    response = client.get("/ers/config/endpoint")
finally:
    client.close()

import httpx

client = httpx.Client(
    base_url="https://ise-01:9060/ers/config",
    timeout=httpx.Timeout(
        connect=5.0,
        read=30.0,
        write=10.0,
        pool=5.0
    ),
    headers={
        "Accept": "application/json",
        "Content-Type": "application/json"
    },
    verify=False,  # SSL verification
    follow_redirects=True,
    http2=True  # Enable HTTP/2
)

# Request with client
response = client.get("/endpoint")  # Full URL: https://ise-01:9060/ers/config/endpoint

import httpx

# Simple
response = httpx.get(
    "https://api.example.com/data",
    auth=("username", "password")
)

# With client
client = httpx.Client(
    base_url="https://ise-01:9060/ers/config",
    auth=("admin", "password")
)

import httpx

# Manual header
response = httpx.get(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer eyJhbGc..."}
)

# Custom auth class
class BearerAuth(httpx.Auth):
    def __init__(self, token: str):
        self.token = token

    def auth_flow(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        yield request

client = httpx.Client(auth=BearerAuth("eyJhbGc..."))

import httpx

# Header-based
client = httpx.Client(
    headers={"X-API-Key": "your-api-key"}
)

# Query parameter
response = httpx.get(
    "https://api.example.com/data",
    params={"api_key": "your-api-key"}
)

import httpx
from datetime import datetime, timedelta

class OAuth2Client:
    def __init__(self, token_url: str, client_id: str, client_secret: str):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token: str | None = None
        self.token_expiry: datetime | None = None

    def get_token(self) -> str:
        """Get valid access token, refreshing if needed."""
        if self._token_valid():
            return self.access_token

        response = httpx.post(
            self.token_url,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        response.raise_for_status()
        data = response.json()

        self.access_token = data["access_token"]
        expires_in = data.get("expires_in", 3600)
        self.token_expiry = datetime.now() + timedelta(seconds=expires_in - 60)

        return self.access_token

    def _token_valid(self) -> bool:
        return (
            self.access_token is not None
            and self.token_expiry is not None
            and datetime.now() < self.token_expiry
        )

# Usage
oauth = OAuth2Client(
    token_url="https://auth.example.com/oauth/token",
    client_id="my-client",
    client_secret="secret"
)

client = httpx.Client(
    headers={"Authorization": f"Bearer {oauth.get_token()}"}
)

import httpx
from dataclasses import dataclass, asdict

@dataclass
class Endpoint:
    mac: str
    group: str
    description: str = ""

class EndpointAPI:
    def __init__(self, base_url: str, auth: tuple[str, str]):
        self.client = httpx.Client(
            base_url=base_url,
            auth=auth,
            headers={"Accept": "application/json", "Content-Type": "application/json"}
        )

    def list(self, page: int = 1, size: int = 100) -> list[dict]:
        """GET /endpoints"""
        response = self.client.get("/endpoints", params={"page": page, "size": size})
        response.raise_for_status()
        return response.json()["resources"]

    def get(self, id: str) -> dict | None:
        """GET /endpoints/{id}"""
        response = self.client.get(f"/endpoints/{id}")
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return response.json()

    def create(self, endpoint: Endpoint) -> str:
        """POST /endpoints"""
        response = self.client.post("/endpoints", json=asdict(endpoint))
        response.raise_for_status()
        # Return ID from Location header
        return response.headers["Location"].split("/")[-1]

    def update(self, id: str, endpoint: Endpoint) -> None:
        """PUT /endpoints/{id}"""
        response = self.client.put(f"/endpoints/{id}", json=asdict(endpoint))
        response.raise_for_status()

    def delete(self, id: str) -> bool:
        """DELETE /endpoints/{id}"""
        response = self.client.delete(f"/endpoints/{id}")
        return response.status_code == 204

    def close(self):
        self.client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

# Usage
with EndpointAPI("https://ise-01:9060/ers/config", ("admin", "password")) as api:
    endpoints = api.list()
    endpoint = api.get("123")
    new_id = api.create(Endpoint(mac="00:11:22:33:44:55", group="Employees"))

import httpx
from typing import Iterator

def paginate(client: httpx.Client, url: str, page_size: int = 100) -> Iterator[dict]:
    """Iterate through paginated API results."""
    page = 1
    while True:
        response = client.get(url, params={"page": page, "size": page_size})
        response.raise_for_status()
        data = response.json()

        resources = data.get("resources", [])
        for item in resources:
            yield item

        # Check if more pages
        total = data.get("total", 0)
        if page * page_size >= total:
            break
        page += 1

# Usage
client = httpx.Client(base_url="https://ise-01:9060/ers/config", auth=("admin", "pass"))
for endpoint in paginate(client, "/endpoint"):
    print(endpoint["mac"])

import httpx

class APIError(Exception):
    def __init__(self, status_code: int, message: str, response: httpx.Response):
        self.status_code = status_code
        self.message = message
        self.response = response
        super().__init__(f"{status_code}: {message}")

class APIClient:
    def __init__(self, base_url: str, auth: tuple[str, str]):
        self.client = httpx.Client(base_url=base_url, auth=auth)

    def request(self, method: str, url: str, **kwargs) -> dict:
        """Make request with error handling."""
        try:
            response = self.client.request(method, url, **kwargs)
        except httpx.ConnectError as e:
            raise APIError(0, f"Connection failed: {e}", None)
        except httpx.TimeoutException as e:
            raise APIError(0, f"Request timed out: {e}", None)

        if response.is_success:
            return response.json() if response.content else {}

        # Parse error message from response
        try:
            error_data = response.json()
            message = error_data.get("message", error_data.get("error", str(error_data)))
        except Exception:
            message = response.text or f"HTTP {response.status_code}"

        raise APIError(response.status_code, message, response)

    def get(self, url: str, **kwargs) -> dict:
        return self.request("GET", url, **kwargs)

    def post(self, url: str, **kwargs) -> dict:
        return self.request("POST", url, **kwargs)

# Usage
client = APIClient("https://ise-01:9060/ers/config", ("admin", "password"))
try:
    data = client.get("/endpoint/invalid-id")
except APIError as e:
    if e.status_code == 404:
        print("Endpoint not found")
    else:
        print(f"API error: {e}")

import httpx
import time
from functools import wraps

def retry(max_attempts: int = 3, backoff: float = 1.0, exceptions: tuple = (httpx.TransportError,)):
    """Decorator for retry with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    if attempt < max_attempts - 1:
                        sleep_time = backoff * (2 ** attempt)
                        time.sleep(sleep_time)
            raise last_error
        return wrapper
    return decorator

class APIClient:
    @retry(max_attempts=3, backoff=1.0)
    def get(self, url: str) -> dict:
        response = self.client.get(url)
        response.raise_for_status()
        return response.json()