Something went wrong. Try again.
Codex skill for Commodore 64 Ultimate hardware workflows.
Something went wrong. Try again.
5.1 kB · 152 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153#!/usr/bin/env python3"""Shared, conservative HTTP helpers for C64 Ultimate REST scripts."""
from __future__ import annotations
import jsonimport timefrom typing import Anyimport urllib.errorimport urllib.request
RETRYABLE_STATUS = frozenset((408, 409, 429))
class UltimateRestError(RuntimeError): """A REST transport or response failure with user-facing detail."""
def host_url(host: str) -> str: """Return a safe HTTP base URL from an explicit host or host:port.""" if not host or any(char.isspace() for char in host) or any( char in host for char in "/@?#" ): raise ValueError("host must be a hostname or IPv4 address, optionally with :port") return f"http://{host}"
def _json_object(payload: bytes, context: str) -> dict[str, Any]: try: decoded = json.loads(payload) if payload else {} except (UnicodeError, json.JSONDecodeError) as error: raise UltimateRestError(f"{context} returned invalid JSON: {error}") from error if not isinstance(decoded, dict): raise UltimateRestError(f"{context} returned non-object JSON") return decoded
def _reported_errors(payload: bytes) -> list[str]: try: decoded = json.loads(payload) if payload else {} except (UnicodeError, json.JSONDecodeError): return [] if not isinstance(decoded, dict): return [] errors = decoded.get("errors", []) if not isinstance(errors, list): return [] return [str(message) for message in errors]
class UltimateRestClient: """Authenticated REST client with bounded, explicit retry behavior."""
def __init__( self, host: str, *, timeout: float, password: str | None, ) -> None: self.base_url = host_url(host) self.timeout = timeout self.password = password
def request( self, route: str, *, method: str = "GET", data: bytes | None = None, content_type: str | None = None, retries: int = 0, retry_404: bool = False, ) -> tuple[bytes, str]: """Return response bytes and content type.
Callers must opt in to retries. Do not enable them for a mutation whose acceptance would be ambiguous after a transport failure. """ if retries < 0: raise ValueError("retries must be non-negative") headers = {"Accept": "application/json, application/octet-stream"} if content_type: headers["Content-Type"] = content_type if self.password: headers["X-Password"] = self.password
for attempt in range(retries + 1): request = urllib.request.Request( self.base_url + route, data=data, headers=headers, method=method, ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return response.read(), response.headers.get_content_type() except urllib.error.HTTPError as error: response_payload = error.read() reported = _reported_errors(response_payload) retryable = ( error.code in RETRYABLE_STATUS or error.code >= 500 or (retry_404 and error.code == 404) ) if retryable and attempt < retries: time.sleep(0.25 * (attempt + 1)) continue detail = f": {'; '.join(reported)}" if reported else "" hint = " (check the network password)" if error.code == 403 else "" raise UltimateRestError( f"{method} {route} HTTP {error.code}: " f"{error.reason}{detail}{hint}" ) from error except OSError as error: if attempt < retries: time.sleep(0.25 * (attempt + 1)) continue raise UltimateRestError( f"{method} {route} transport failed: {error}" ) from error
raise UltimateRestError(f"{method} {route} exhausted its retry budget")
def json( self, route: str, *, method: str = "GET", data: bytes | None = None, content_type: str | None = None, retries: int = 0, retry_404: bool = False, ) -> dict[str, Any]: payload, _ = self.request( route, method=method, data=data, content_type=content_type, retries=retries, retry_404=retry_404, ) decoded = _json_object(payload, f"{method} {route}") errors = decoded.get("errors", []) if errors: if isinstance(errors, list): detail = "; ".join(map(str, errors)) else: detail = str(errors) raise UltimateRestError(f"{method} {route}: {detail}") return decoded