python_client.py

"""
Python client for shared-hosting-deploy-kit.

Usage:

    from deploy_client import Gateway

    gw = Gateway("https://example.com/deploy-gateway.php", api_key="...")

    # Conflict-safe read-edit-write
    current = gw.read("index.html")
    new = current["content"].replace("foo", "bar")
    gw.write("index.html", new, if_match=current["sha256"])

License: MIT
"""

from __future__ import annotations

import json
import time
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import HTTPError


class GatewayError(Exception):
    """Generic gateway error (4xx/5xx that is not 409)."""


class ConflictError(GatewayError):
    """Raised on 409 Conflict (If-Match precondition failed)."""

    def __init__(self, expected: str, current: str, path: str):
        super().__init__(
            f"If-Match conflict on {path!r}: expected {expected}, got {current}"
        )
        self.expected = expected
        self.current = current
        self.path = path


@dataclass
class Gateway:
    url: str
    api_key: str
    timeout: int = 30
    min_sleep_between_calls: float = 0.0  # set to 5-7 for tight rate limits

    _last_call: float = 0.0

    def _post(self, action: str, **fields) -> dict:
        if self.min_sleep_between_calls > 0:
            delta = time.time() - self._last_call
            if delta < self.min_sleep_between_calls:
                time.sleep(self.min_sleep_between_calls - delta)
        payload = {"key": self.api_key, "action": action, **fields}
        data = urlencode(payload).encode("utf-8")
        req = Request(self.url, data=data, method="POST")
        try:
            with urlopen(req, timeout=self.timeout) as resp:
                raw = resp.read().decode("utf-8", errors="replace")
        except HTTPError as e:
            raw = e.read().decode("utf-8", errors="replace")
            try:
                body = json.loads(raw, strict=False)
            except json.JSONDecodeError:
                body = {"error": raw[:200]}
            if e.code == 409:
                raise ConflictError(
                    expected=body.get("expected_sha256", ""),
                    current=body.get("current_sha256", ""),
                    path=body.get("path", fields.get("path", "")),
                )
            raise GatewayError(f"HTTP {e.code}: {body.get('error', raw[:200])}")
        finally:
            self._last_call = time.time()

        try:
            return json.loads(raw, strict=False)
        except json.JSONDecodeError as e:
            raise GatewayError(f"Invalid JSON from gateway: {e}; raw={raw[:200]!r}")

    # ---- High-level API ----

    def ping(self) -> dict:
        return self._post("ping")

    def read(self, path: str) -> dict:
        """Return {'content', 'size', 'sha256', 'mtime'}."""
        return self._post("read", path=path)

    def write(
        self,
        path: str,
        content: str,
        if_match: Optional[str] = None,
    ) -> dict:
        """Write a text file. Pass if_match=sha256 from a previous read for safe edits."""
        fields = {"path": path, "content": content}
        if if_match:
            fields["if_match"] = if_match
        return self._post("write", **fields)

    def write_binary(
        self,
        path: str,
        data: bytes,
        if_match: Optional[str] = None,
    ) -> dict:
        """Upload binary data via base64. For large files (>800 KB) use bin-upload.php instead."""
        import base64
        fields = {"path": path, "content": base64.b64encode(data).decode("ascii")}
        if if_match:
            fields["if_match"] = if_match
        return self._post("write_b64", **fields)

    def delete(self, path: str) -> dict:
        return self._post("delete", path=path)

    def list(self, path: str = ".") -> list[dict]:
        return self._post("list", path=path).get("items", [])

    def snapshot(self, label: str = "manual") -> dict:
        return self._post("snapshot", label=label).get("snapshot", {})

    def snapshots(self) -> list[dict]:
        return self._post("snapshots").get("snapshots", [])

    def rollback(self, snapshot_id: str) -> dict:
        return self._post("rollback", snapshot_id=snapshot_id)


# ---- Demo ----

if __name__ == "__main__":
    import os
    import sys

    url = os.environ.get("DEPLOY_GATEWAY_URL")
    key = os.environ.get("DEPLOY_GATEWAY_KEY")
    if not url or not key:
        sys.exit("Set DEPLOY_GATEWAY_URL and DEPLOY_GATEWAY_KEY env vars first.")

    gw = Gateway(url, key)
    print("ping:", gw.ping())
    print("snapshots:", len(gw.snapshots()))

    # Round-trip example
    test_path = "test-deploy-kit.txt"
    gw.write(test_path, f"hello from python at {time.time()}")
    info = gw.read(test_path)
    print(f"wrote {info['size']} bytes, sha={info['sha256'][:12]}...")
    gw.delete(test_path)
    print("cleaned up")