homelab-codex-ws/scripts/ha/lib/ha_write_api.py

#!/usr/bin/env python3
"""Write-capable REST client for the HA "api" adapter's deploy path (see
kb/decisions/ha-configs-as-code.md, "Sync model", "Validation gate").

Deliberately kept out of ha_api.py, whose module docstring guarantees that
module is read-only by construction — the import path relies on that
guarantee staying true. This module is the only place under scripts/ha/
that ever issues a mutating call to a live HA instance, and it exposes only
the two operations DESIGN.md's deploy sequence requires:

  - check_config()          -> POST /api/config/core/check_config (validation gate)
  - post_config(...)        -> POST /api/config/<domain>/config/<id> (per-object write)

Same token discipline as ha_api.py: callers pass a token_path (via
ha_api.read_token), never a raw token in argv or a log line; the
Authorization header is set in-process via `requests`.
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import requests  # noqa: E402

import ha_api  # noqa: E402


class WriteClient(ha_api.Client):
    """Extends the read-only Client with the two mutating calls deploy.sh needs."""

    def post_config(self, domain, object_id, body):
        """POST `body` (a dict) to /api/config/<domain>/config/<object_id>.

        Returns the parsed JSON response. Raises requests.HTTPError on a
        non-2xx status — deploy_api.py treats a failed write as a
        per-object error, never a silent skip.
        """
        path = f"/api/config/{domain}/config/{object_id}"
        resp = requests.post(
            self._base_url + path, headers=self._headers(), json=body, timeout=self._timeout
        )
        resp.raise_for_status()
        return resp.json()

    def check_config(self):
        """POST /api/config/core/check_config — HA's built-in config validator.

        Returns (valid: bool, raw: dict). Only raises on a transport/HTTP
        failure; an "invalid" result is a normal outcome the caller gates
        on, not an exception.
        """
        resp = requests.post(
            self._base_url + "/api/config/core/check_config",
            headers=self._headers(),
            timeout=self._timeout,
        )
        resp.raise_for_status()
        data = resp.json()
        return data.get("result") == "valid", data