96fa96c61e
Keep FastAPI as the only backend; add media upload, ntfy hook, and PB collection bootstrap for production. Co-authored-by: Cursor <cursoragent@cursor.com>
81 lines
1.8 KiB
Python
81 lines
1.8 KiB
Python
"""Shared PocketBase helpers for store modules."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.services.pb_client import PocketBaseError, pb, pb_quote
|
|
|
|
_pb_ready: bool | None = None
|
|
|
|
|
|
def use_pb() -> bool:
|
|
global _pb_ready
|
|
if _pb_ready is not None:
|
|
return _pb_ready
|
|
try:
|
|
if not pb.health_ok():
|
|
_pb_ready = False
|
|
return False
|
|
pb.admin_token()
|
|
_pb_ready = True
|
|
except Exception:
|
|
_pb_ready = False
|
|
return _pb_ready
|
|
|
|
|
|
def reset_pb_cache() -> None:
|
|
global _pb_ready
|
|
_pb_ready = None
|
|
|
|
|
|
def q(value: str) -> str:
|
|
return pb_quote(value)
|
|
|
|
|
|
def rid(record: dict[str, Any], legacy_field: str = "legacyId") -> str:
|
|
return str(record.get(legacy_field) or record.get("id") or "")
|
|
|
|
|
|
def pb_date(value: Any) -> str:
|
|
if not value:
|
|
return ""
|
|
s = str(value)
|
|
return s[:10] if len(s) >= 10 else s
|
|
|
|
|
|
def safe_list(collection: str, **kwargs: Any) -> list[dict[str, Any]]:
|
|
try:
|
|
return pb.list_all(collection, **kwargs)
|
|
except PocketBaseError:
|
|
return []
|
|
|
|
|
|
def safe_first(collection: str, *, filter: str) -> dict[str, Any] | None:
|
|
try:
|
|
return pb.first_record(collection, filter=filter)
|
|
except PocketBaseError:
|
|
return None
|
|
|
|
|
|
def safe_create(collection: str, payload: dict[str, Any]) -> dict[str, Any] | None:
|
|
try:
|
|
return pb.create_record(collection, payload)
|
|
except PocketBaseError:
|
|
return None
|
|
|
|
|
|
def safe_update(collection: str, record_id: str, payload: dict[str, Any]) -> dict[str, Any] | None:
|
|
try:
|
|
return pb.update_record(collection, record_id, payload)
|
|
except PocketBaseError:
|
|
return None
|
|
|
|
|
|
def find_by_legacy(collection: str, legacy_id: str) -> dict[str, Any] | None:
|
|
if not legacy_id:
|
|
return None
|
|
row = safe_first(collection, filter=f"legacyId={q(legacy_id)}")
|
|
if row:
|
|
return row
|
|
return safe_first(collection, filter=f"id={q(legacy_id)}")
|