envelope.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """JSON response envelope for hosted-dashboard ``api/*`` (§HGD.5.3).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from dataclasses import dataclass |
| 7 | from datetime import datetime, timezone |
| 8 | from typing import Any |
| 9 | |
| 10 | |
| 11 | def utc_now_iso() -> str: |
| 12 | """Return current UTC timestamp in ISO-8601 with ``Z`` suffix.""" |
| 13 | return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
| 14 | |
| 15 | |
| 16 | @dataclass(frozen=True) |
| 17 | class ApiEnvelope: |
| 18 | """Standard hosted-dashboard API response wrapper.""" |
| 19 | |
| 20 | ok: bool |
| 21 | result: Any |
| 22 | error: str | None = None |
| 23 | http_status: int = 200 |
| 24 | meta: dict[str, Any] | None = None |
| 25 | |
| 26 | def to_dict(self) -> dict[str, Any]: |
| 27 | payload: dict[str, Any] = { |
| 28 | "ok": self.ok, |
| 29 | "result": self.result, |
| 30 | } |
| 31 | if self.error is not None: |
| 32 | payload["error"] = self.error |
| 33 | if self.meta is not None: |
| 34 | payload["meta"] = self.meta |
| 35 | return payload |
| 36 | |
| 37 | def to_json_bytes(self) -> bytes: |
| 38 | return json.dumps(self.to_dict(), indent=2, sort_keys=True).encode("utf-8") |
| 39 | |
| 40 | |
| 41 | def build_meta( |
| 42 | *, |
| 43 | source_id: str, |
| 44 | ref: str, |
| 45 | content_sha256: str | None = None, |
| 46 | fetched_at: str | None = None, |
| 47 | ) -> dict[str, Any]: |
| 48 | """Build the frozen ``meta`` object; ``authoritative_workflow`` is always ``local``.""" |
| 49 | meta: dict[str, Any] = { |
| 50 | "source_id": source_id, |
| 51 | "ref": ref, |
| 52 | "fetched_at": fetched_at or utc_now_iso(), |
| 53 | "authoritative_workflow": "local", |
| 54 | } |
| 55 | if content_sha256 is not None: |
| 56 | meta["content_sha256"] = content_sha256 |
| 57 | return meta |
| 58 | |
| 59 | |
| 60 | def success(result: Any, *, meta: dict[str, Any], http_status: int = 200) -> ApiEnvelope: |
| 61 | """Successful response with required meta.""" |
| 62 | return ApiEnvelope(ok=True, result=result, error=None, http_status=http_status, meta=meta) |
| 63 | |
| 64 | |
| 65 | def failure( |
| 66 | *, |
| 67 | error: str, |
| 68 | http_status: int, |
| 69 | result: Any = None, |
| 70 | meta: dict[str, Any] | None = None, |
| 71 | ) -> ApiEnvelope: |
| 72 | """Failed response with error token.""" |
| 73 | return ApiEnvelope(ok=False, result=result, error=error, http_status=http_status, meta=meta) |
| 74 | |
| 75 | |
| 76 | def health_success() -> ApiEnvelope: |
| 77 | """``api/health`` body (§HGD.5.4).""" |
| 78 | return ApiEnvelope( |
| 79 | ok=True, |
| 80 | result={"status": "ok", "mode": "hosted-read-only"}, |
| 81 | error=None, |
| 82 | http_status=200, |
| 83 | meta=build_meta(source_id="github_meta", ref="n/a", content_sha256=None), |
| 84 | ) |