nomadweb/backend/app/services/pb_client.py

"""PocketBase admin client — full CRUD for FastAPI services."""
from __future__ import annotations

from typing import Any
from urllib.parse import quote

import httpx

from app.config import settings


class PocketBaseError(RuntimeError):
  pass


def pb_quote(value: str) -> str:
  return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'


class PocketBaseClient:
  def __init__(self) -> None:
    self._admin_token: str | None = None
    self._client: httpx.Client | None = None

  @property
  def base_url(self) -> str:
    return settings.pocketbase_url.rstrip("/")

  def _http(self) -> httpx.Client:
    if self._client is None:
      self._client = httpx.Client(timeout=20.0, trust_env=False)
    return self._client

  def request(
    self,
    method: str,
    path: str,
    *,
    token: str | None = None,
    admin: bool = False,
    json: dict[str, Any] | None = None,
    params: dict[str, Any] | None = None,
  ) -> Any:
    headers: dict[str, str] = {}
    if admin:
      token = self.admin_token()
    if token:
      headers["Authorization"] = token if token.startswith("Bearer ") else token
    url = f"{self.base_url}{path}"
    res = self._http().request(method, url, headers=headers, json=json, params=params)
    if res.status_code >= 400:
      try:
        detail = res.json()
      except Exception:
        detail = res.text
      raise PocketBaseError(f"PocketBase {res.status_code}: {detail}")
    if not res.content:
      return None
    return res.json()

  def admin_token(self) -> str:
    if self._admin_token:
      return self._admin_token
    body = {
      "identity": settings.pocketbase_admin_email,
      "password": settings.pocketbase_admin_password,
    }
    last_error: Exception | None = None
    for endpoint in (
      "/api/collections/_superusers/auth-with-password",
      "/api/admins/auth-with-password",
    ):
      try:
        data = self.request("POST", endpoint, json=body)
        token = (data or {}).get("token")
        if token:
          self._admin_token = token
          return token
      except Exception as exc:
        last_error = exc
    raise PocketBaseError(f"Unable to authenticate PocketBase admin: {last_error}")

  def list_records(
    self,
    collection: str,
    *,
    filter: str | None = None,
    sort: str | None = None,
    page: int = 1,
    per_page: int = 200,
    admin: bool = True,
  ) -> dict[str, Any]:
    params: dict[str, Any] = {"page": page, "perPage": per_page}
    if filter:
      params["filter"] = filter
    if sort:
      params["sort"] = sort
    return self.request(
      "GET",
      f"/api/collections/{collection}/records",
      admin=admin,
      params=params,
    )

  def list_all(self, collection: str, *, filter: str | None = None, sort: str | None = None) -> list[dict[str, Any]]:
    page = 1
    items: list[dict[str, Any]] = []
    while True:
      data = self.list_records(collection, filter=filter, sort=sort, page=page)
      batch = data.get("items") or []
      items.extend(batch)
      if page >= (data.get("totalPages") or 1):
        break
      page += 1
    return items

  def first_record(self, collection: str, *, filter: str, admin: bool = True) -> dict[str, Any] | None:
    data = self.list_records(collection, filter=filter, per_page=1, admin=admin)
    items = data.get("items") or []
    return items[0] if items else None

  def create_record(self, collection: str, payload: dict[str, Any]) -> dict[str, Any]:
    return self.request(
      "POST",
      f"/api/collections/{collection}/records",
      admin=True,
      json=payload,
    )

  def update_record(self, collection: str, record_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    return self.request(
      "PATCH",
      f"/api/collections/{collection}/records/{record_id}",
      admin=True,
      json=payload,
    )

  def delete_record(self, collection: str, record_id: str) -> None:
    self.request(
      "DELETE",
      f"/api/collections/{collection}/records/{record_id}",
      admin=True,
    )

  def upsert_by_field(
    self,
    collection: str,
    field: str,
    value: str,
    payload: dict[str, Any],
  ) -> dict[str, Any]:
    existing = self.first_record(collection, filter=f"{field}={pb_quote(value)}")
    if existing:
      return self.update_record(collection, existing["id"], payload)
    return self.create_record(collection, payload)

  def health_ok(self) -> bool:
    try:
      res = self._http().get(f"{self.base_url}/api/health", timeout=2.0)
      return res.status_code == 200
    except Exception:
      return False


pb = PocketBaseClient()