159 lines
4.5 KiB
Python
159 lines
4.5 KiB
Python
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()
|