Spaces:
Sleeping
Sleeping
| """Thin client for the Supabase REST API (PostgREST). | |
| Only the server holds the service_role key. Filter values passed here are | |
| always server-generated (validated UUIDs or fixed status words), never raw | |
| user input; searching and filtering of listings happens in Python. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import uuid | |
| from typing import Any, Optional | |
| import httpx | |
| log = logging.getLogger("curas.db") | |
| class DbError(Exception): | |
| def __init__(self, message: str, status: Optional[int] = None): | |
| super().__init__(message) | |
| self.status = status | |
| class DbUnavailable(DbError): | |
| pass | |
| def check_uuid(value: str) -> str: | |
| """Return the canonical form of a UUID string or raise ValueError.""" | |
| return str(uuid.UUID(str(value))) | |
| class Db: | |
| def __init__(self, url: str, key: str, rest_path: str = "/rest/v1", timeout: float = 10.0): | |
| if not url or not key: | |
| raise DbUnavailable("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY are not configured") | |
| self.base = url.rstrip("/") + (rest_path if rest_path.startswith("/") or rest_path == "" else "/" + rest_path) | |
| # New-style Supabase secret keys (sb_secret_...) are sent only on the | |
| # apikey header; legacy service_role JWTs are sent on both. | |
| self._headers = {"apikey": key, "Accept": "application/json"} | |
| if not key.startswith("sb_"): | |
| self._headers["Authorization"] = f"Bearer {key}" | |
| self._client = httpx.Client(timeout=timeout, headers=self._headers) | |
| def close(self) -> None: | |
| self._client.close() | |
| # -- low level --------------------------------------------------------- | |
| def _request(self, method: str, table: str, *, params: Optional[dict] = None, | |
| json: Any = None, prefer: Optional[str] = None) -> Any: | |
| headers = {} | |
| if prefer: | |
| headers["Prefer"] = prefer | |
| if json is not None: | |
| headers["Content-Type"] = "application/json" | |
| try: | |
| resp = self._send(method, table, params, json, headers) | |
| except httpx.HTTPError as exc: | |
| log.error("database request failed: %s %s: %s", method, table, exc) | |
| raise DbUnavailable(f"database request failed: {exc.__class__.__name__}") from exc | |
| if resp.status_code >= 400: | |
| detail = resp.text[:500] | |
| log.error("database error %s on %s %s: %s", resp.status_code, method, table, detail) | |
| raise DbError(f"database error {resp.status_code}", resp.status_code) | |
| if resp.status_code == 204 or not resp.content: | |
| return None | |
| return resp.json() | |
| def _send(self, method: str, table: str, params, json, headers): | |
| """Send a request; retry a read once if a pooled connection was dropped.""" | |
| try: | |
| return self._client.request(method, f"{self.base}/{table}", params=params, | |
| json=json, headers=headers) | |
| except (httpx.RemoteProtocolError, httpx.ConnectError): | |
| if method not in ("GET", "HEAD"): | |
| raise | |
| return self._client.request(method, f"{self.base}/{table}", params=params, | |
| json=json, headers=headers) | |
| # -- helpers ----------------------------------------------------------- | |
| def select(self, table: str, filters: Optional[dict] = None, *, order: Optional[str] = None, | |
| columns: str = "*", limit: Optional[int] = None) -> list[dict]: | |
| params: dict[str, str] = {"select": columns} | |
| params.update(filters or {}) | |
| if order: | |
| params["order"] = order | |
| if limit is not None: | |
| params["limit"] = str(limit) | |
| return self._request("GET", table, params=params) or [] | |
| def get_one(self, table: str, id_: str) -> Optional[dict]: | |
| rows = self.select(table, {"id": f"eq.{check_uuid(id_)}"}, limit=1) | |
| return rows[0] if rows else None | |
| def insert(self, table: str, row: dict) -> dict: | |
| rows = self._request("POST", table, json=row, prefer="return=representation") | |
| if not rows: | |
| raise DbError("insert returned no row") | |
| return rows[0] | |
| def insert_many(self, table: str, rows: list[dict]) -> list[dict]: | |
| if not rows: | |
| return [] | |
| return self._request("POST", table, json=rows, prefer="return=representation") or [] | |
| def update(self, table: str, filters: dict, patch: dict) -> list[dict]: | |
| """PATCH rows matching filters. Returns the updated rows (may be empty).""" | |
| return self._request("PATCH", table, params=filters, json=patch, | |
| prefer="return=representation") or [] | |
| def count(self, table: str, filters: Optional[dict] = None) -> int: | |
| """Exact row count via a zero-width range request (keeps the pooled | |
| connection reusable, unlike HEAD).""" | |
| params: dict[str, str] = {"select": "id"} | |
| params.update(filters or {}) | |
| headers = {"Prefer": "count=exact", "Range-Unit": "items", "Range": "0-0"} | |
| try: | |
| resp = self._send("GET", table, params, None, headers) | |
| except httpx.HTTPError as exc: | |
| raise DbUnavailable(f"database request failed: {exc.__class__.__name__}") from exc | |
| if resp.status_code >= 400: | |
| raise DbError(f"database error {resp.status_code}", resp.status_code) | |
| rng = resp.headers.get("content-range", "") | |
| try: | |
| return int(rng.split("/")[-1]) | |
| except ValueError: | |
| return 0 | |