58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class SupabaseRestClient:
|
|
"""轻量封装 Supabase RESTful API，兼容本地 sb_secret 密钥。"""
|
|
|
|
def __init__(self) -> None:
|
|
base_url = settings.supabase_url.rstrip("/")
|
|
self.client = httpx.Client(
|
|
base_url=f"{base_url}/rest/v1",
|
|
headers={
|
|
"apikey": settings.supabase_service_role_key,
|
|
"Authorization": f"Bearer {settings.supabase_service_role_key}",
|
|
},
|
|
timeout=10.0,
|
|
)
|
|
|
|
def insert(self, table: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
response = self.client.post(
|
|
f"/{table}",
|
|
json=payload,
|
|
headers={"Prefer": "return=representation"},
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if isinstance(data, list):
|
|
return data[0]
|
|
return data
|
|
|
|
def select_one(self, table: str, filters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
params = {key: f"eq.{value}" for key, value in filters.items()}
|
|
params["select"] = "*"
|
|
response = self.client.get(f"/{table}", params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if isinstance(data, list) and data:
|
|
return data[0]
|
|
return None
|
|
|
|
def update(self, table: str, filters: Dict[str, Any], payload: Dict[str, Any]) -> None:
|
|
params = {key: f"eq.{value}" for key, value in filters.items()}
|
|
response = self.client.patch(
|
|
f"/{table}",
|
|
params=params,
|
|
json=payload,
|
|
headers={"Prefer": "return=minimal"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
|
|
supabase_rest = SupabaseRestClient()
|