moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(datastore): migrate to single-writer DuckDB backend HTTP proxy and remove legacy HTML renderers

Browse Source
This commit is contained in:
Moritz Graf
2026-10-16 23:33:02 +02:00
parent 8bdd06e598
commit 6b3fbfdbb5
6 changed files with 184 additions and 3688 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/datastore/scripts/datastore.py
+124 -29
View File
@@ -29,6 +29,8 @@ from typing import Any, Dict, List, Optional, Tuple, Union
from zoneinfo import ZoneInfo
import duckdb
import urllib.request
import urllib.error
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
@@ -110,12 +112,74 @@ def find_data_dir() -> Path:
class DatastoreEngine:
def __init__(self, data_dir: Optional[Path] = None):
def __init__(self, data_dir: Optional[Path] = None, backend_url: Optional[str] = None):
self.data_dir = data_dir or find_data_dir()
self.con = duckdb.connect(":memory:")
self.has_spatial_extension = False
self._init_schemas()
self._load_all_data()
self.backend_url = (backend_url or os.environ.get(
"DUCKDB_BACKEND_URL", "http://duckdb-backend.family.svc.cluster.local:8000/duckdb-api/v1"
)).rstrip("/")
self.use_backend = False
if os.environ.get("DUCKDB_FORCE_LOCAL") != "1":
try:
base = self.backend_url.replace("/duckdb-api/v1", "")
test_url = f"{base}/healthz"
req = urllib.request.Request(test_url, method="GET")
with urllib.request.urlopen(req, timeout=1.0) as r:
if r.status == 200:
self.use_backend = True
logger.info(f"DatastoreEngine using single-writer backend at {self.backend_url}")
except Exception:
self.use_backend = False
if not self.use_backend:
self.con = duckdb.connect(":memory:")
self.has_spatial_extension = False
self._init_schemas()
self._load_all_data()
def _http_post(self, url: str, data: Dict[str, Any]) -> Any:
req = urllib.request.Request(
url,
data=json.dumps(data).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
try:
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = err_body
raise RuntimeError(f"Backend HTTP {e.code}: {detail}")
def _http_get(self, url: str) -> Any:
req = urllib.request.Request(url, method="GET")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
try:
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = err_body
raise RuntimeError(f"Backend HTTP {e.code}: {detail}")
def _http_delete(self, url: str) -> Any:
req = urllib.request.Request(url, method="DELETE")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
try:
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = err_body
raise RuntimeError(f"Backend HTTP {e.code}: {detail}")
def _init_schemas(self):
"""Initializes tables, views, and spatial support in DuckDB."""
@@ -429,30 +493,8 @@ class DatastoreEngine:
logger.warning(f"Error loading garmin record from {file_path}: {e}")
def _trigger_website_update(self, subject: str):
"""Triggers website updates automatically (Mila -> update_website.py, Moritz -> update_moritz_website.py)."""
if subject == "mila":
script_name = "update_website.py"
elif subject == "moritz":
script_name = "update_moritz_website.py"
else:
return
try:
update_script = Path(__file__).resolve().parent / script_name
if update_script.exists():
import subprocess, shutil
uv_bin = shutil.which("uv") or "/home/node/.openclaw/bin/uv"
if Path(uv_bin).is_file():
cmd = [str(uv_bin), "run", str(update_script), "--data-dir", str(self.data_dir)]
else:
cmd = [sys.executable, str(update_script), "--data-dir", str(self.data_dir)]
subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception as e:
logger.warning(f"Could not trigger {script_name}: {e}")
"""No-op: legacy HTML generator retired in favor of Vue 3 apps."""
pass
def _git_sync(self, commit_msg: str):
"""Safely commits and pushes data changes if data_dir is inside a git repository."""
@@ -653,6 +695,25 @@ class DatastoreEngine:
allow_duplicate: bool = False,
) -> Dict[str, Any]:
"""Adds a time-series event into the appropriate database (sleep, nutrition, or mila)."""
if self.use_backend:
body = {
"kind": kind,
"subject": subject,
"actor": actor,
"payload": payload,
"start": start,
"end": end,
"since": since,
"estimated": estimated,
"meal": meal,
"note": note,
"source_ref": source_ref,
"allow_duplicate": allow_duplicate,
}
if isinstance(payload, dict) and "date" in payload and payload["date"]:
body["date"] = payload["date"]
return self._http_post(f"{self.backend_url}/events", body)
event_id = generate_ulid()
now_dt = datetime.datetime.now(datetime.timezone.utc)
now_iso = now_dt.isoformat()
@@ -905,6 +966,18 @@ class DatastoreEngine:
actor: Optional[str] = None,
) -> Dict[str, Any]:
"""Finds the open event (end IS NULL) for kind and subject, and closes it."""
if self.use_backend:
body = {
"kind": kind,
"subject": subject,
"end_time": end_time,
"since": since,
"event_id": event_id,
}
if actor:
body["actor"] = actor
return self._http_post(f"{self.backend_url}/events/stop", body)
if since and end_time:
raise ValueError("Cannot specify both 'since' and 'end_time'. Use 'since' for relative times or 'end_time' for absolute times.")
@@ -1030,6 +1103,9 @@ class DatastoreEngine:
def delete_event(self, event_id: str) -> Dict[str, Any]:
"""Soft-deletes an event across sleep, nutrition, or mila."""
if self.use_backend:
return self._http_delete(f"{self.backend_url}/events/{event_id}")
now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
row = self.con.execute("SELECT kind, subject FROM events WHERE id = ?", [event_id]).fetchone()
if not row:
@@ -1052,6 +1128,9 @@ class DatastoreEngine:
# --------------------------------------------------------------------------
def put_entity(self, collection: str, record: Dict[str, Any]) -> Dict[str, Any]:
"""Upserts an entity into entities/<collection>.jsonl."""
if self.use_backend:
return self._http_post(f"{self.backend_url}/entities", {"collection": collection, "record": record})
if collection not in ["foods", "recipes", "places", "people"]:
raise ValueError(f"Unknown entity collection: {collection}")
@@ -1114,6 +1193,10 @@ class DatastoreEngine:
# --------------------------------------------------------------------------
def query(self, sql: str) -> List[Dict[str, Any]]:
"""Executes read-only SQL query and returns list of dictionaries."""
if self.use_backend:
res = self._http_post(f"{self.backend_url}/query", {"query": sql})
return res.get("rows", [])
clean_sql = sql.strip()
# Basic safeguard against destructive statements in read query
disallowed = ["DROP ", "ALTER ", "TRUNCATE ", "DELETE FROM", "UPDATE "]
@@ -1138,6 +1221,12 @@ class DatastoreEngine:
def get_summary(self, subject: str, date_str: Optional[str] = None) -> Dict[str, Any]:
"""Returns daily summary for subject (nutrition, sleep, deficit)."""
if self.use_backend:
url = f"{self.backend_url}/summary/{subject}"
if date_str:
url += f"?date={date_str}"
return self._http_get(url)
target_date = date_str or datetime.datetime.now(BERLIN_TZ).strftime("%Y-%m-%d")
res: Dict[str, Any] = {"subject": subject, "date": target_date}
@@ -1195,6 +1284,12 @@ class DatastoreEngine:
category: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""Radius search for places using spatial extension or Haversine fallback."""
if self.use_backend:
url = f"{self.backend_url}/places/near?lat={lat}&lon={lon}&radius_m={radius_m}"
if category:
url += f"&category={category}"
return self._http_get(url)
cat_filter = f"AND category = '{category}'" if category else ""
if self.has_spatial_extension: