| from __future__ import annotations |
|
|
| import json |
| import urllib.error |
| import urllib.request |
| from typing import Any |
|
|
| from time_machine.domain.errors import AdapterConfigurationError |
|
|
|
|
| class ModalEndpointClient: |
| def __init__( |
| self, |
| url: str, |
| timeout_seconds: float = 120.0, |
| bearer_token: str | None = None, |
| ) -> None: |
| clean_url = url.strip() |
| if not clean_url: |
| raise AdapterConfigurationError("Modal endpoint URL must not be empty.") |
| self.url = clean_url |
| self.timeout_seconds = timeout_seconds |
| self.bearer_token = bearer_token.strip() if bearer_token else None |
|
|
| def post_json(self, payload: dict[str, Any]) -> dict[str, Any]: |
| body = json.dumps(payload).encode("utf-8") |
| headers = { |
| "Content-Type": "application/json", |
| "Accept": "application/json", |
| } |
| if self.bearer_token: |
| headers["Authorization"] = f"Bearer {self.bearer_token}" |
|
|
| request = urllib.request.Request( |
| self.url, |
| data=body, |
| headers=headers, |
| method="POST", |
| ) |
| try: |
| with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: |
| response_body = response.read().decode("utf-8") |
| except urllib.error.HTTPError as exc: |
| detail = exc.read().decode("utf-8", errors="replace") |
| raise RuntimeError( |
| f"Modal endpoint {self.url} returned HTTP {exc.code}: {detail}" |
| ) from exc |
| except urllib.error.URLError as exc: |
| raise RuntimeError(f"Modal endpoint {self.url} is not reachable: {exc}") from exc |
|
|
| try: |
| decoded = json.loads(response_body) |
| except json.JSONDecodeError as exc: |
| raise RuntimeError( |
| f"Modal endpoint {self.url} returned non-JSON response." |
| ) from exc |
| if not isinstance(decoded, dict): |
| raise RuntimeError(f"Modal endpoint {self.url} returned a non-object JSON value.") |
| return decoded |
|
|
