"""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()