s''

backend/pb_client.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urlencode
+
+import httpx
+
+from .settings import get_settings
+
+
+class PocketBaseError(RuntimeError):
+    pass
+
+
+def pb_quote(value: str) -> str:
+    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+class PocketBaseClient:
+    def __init__(self) -> None:
+        self.settings = get_settings()
+        self._admin_token: str | None = None
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def base_url(self) -> str:
+        return self.settings.pb_url
+
+    async def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        token: str | None = None,
+        admin: bool = False,
+        json: dict[str, Any] | None = None,
+        data: dict[str, Any] | None = None,
+        files: dict[str, Any] | None = None,
+        params: dict[str, Any] | None = None,
+    ) -> Any:
+        headers: dict[str, str] = {}
+        if admin:
+            token = await self.admin_token()
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+        url = f"{self.base_url}{path}"
+        if self._client is None:
+            self._client = httpx.AsyncClient(timeout=20.0, trust_env=False)
+        res = await self._client.request(
+            method,
+            url,
+            headers=headers,
+            json=json,
+            data=data,
+            files=files,
+            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()
+
+    async def admin_token(self) -> str:
+        if self._admin_token:
+            return self._admin_token
+        body = {
+            "identity": self.settings.pb_admin_email,
+            "password": self.settings.pb_admin_password,
+        }
+        endpoints = [
+            "/api/collections/_superusers/auth-with-password",
+            "/api/admins/auth-with-password",
+        ]
+        last_error: Exception | None = None
+        for endpoint in endpoints:
+            try:
+                data = await self.request("POST", endpoint, json=body)
+                token = data.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}")
+
+    async def list_records(
+        self,
+        collection: str,
+        *,
+        filter: str | None = None,
+        sort: str | None = None,
+        page: int = 1,
+        per_page: int = 50,
+        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 await self.request(
+            "GET",
+            f"/api/collections/{collection}/records",
+            admin=admin,
+            params=params,
+        )
+
+    async def first_record(
+        self,
+        collection: str,
+        *,
+        filter: str,
+        admin: bool = True,
+    ) -> dict[str, Any] | None:
+        data = await self.list_records(collection, filter=filter, per_page=1, admin=admin)
+        items = data.get("items") or []
+        return items[0] if items else None
+
+    async def create_record(self, collection: str, payload: dict[str, Any]) -> dict[str, Any]:
+        return await self.request(
+            "POST",
+            f"/api/collections/{collection}/records",
+            admin=True,
+            json=payload,
+        )
+
+    async def update_record(
+        self,
+        collection: str,
+        record_id: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        return await self.request(
+            "PATCH",
+            f"/api/collections/{collection}/records/{record_id}",
+            admin=True,
+            json=payload,
+        )
+
+    async def upsert_by_field(
+        self,
+        collection: str,
+        field: str,
+        value: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        existing = await self.first_record(collection, filter=f"{field}={pb_quote(value)}")
+        if existing:
+            return await self.update_record(collection, existing["id"], payload)
+        return await self.create_record(collection, payload)
+
+
+pb = PocketBaseClient()