Add Doris Schoolhouse and clean pending homelab changes
This commit is contained in:
53
home/doris-schoolhouse/app/services/store.py
Normal file
53
home/doris-schoolhouse/app/services/store.py
Normal file
@@ -0,0 +1,53 @@
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
from app.config import settings
|
||||
|
||||
|
||||
class JsonStore:
|
||||
def __init__(self) -> None:
|
||||
self.root = Path(settings.state_dir)
|
||||
self.root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _path(self, name: str) -> Path:
|
||||
return self.root / f"{name}.json"
|
||||
|
||||
def load(self, name: str, default: Any) -> Any:
|
||||
path = self._path(name)
|
||||
if not path.exists():
|
||||
return default
|
||||
return json.loads(path.read_text())
|
||||
|
||||
def save(self, name: str, value: Any) -> Any:
|
||||
path = self._path(name)
|
||||
path.write_text(json.dumps(value, indent=2, sort_keys=True))
|
||||
return value
|
||||
|
||||
def replace_collection(self, name: str, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
return self.save(name, items)
|
||||
|
||||
def append_collection(self, name: str, item: dict[str, Any]) -> dict[str, Any]:
|
||||
items = self.load(name, [])
|
||||
items.append(item)
|
||||
self.save(name, items)
|
||||
return item
|
||||
|
||||
def update_collection_item(self, name: str, predicate, updater):
|
||||
items = self.load(name, [])
|
||||
updated = None
|
||||
for idx, item in enumerate(items):
|
||||
if predicate(item):
|
||||
items[idx] = updater(dict(item))
|
||||
updated = items[idx]
|
||||
break
|
||||
self.save(name, items)
|
||||
return updated
|
||||
|
||||
def now(self) -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
def new_id(self, prefix: str) -> str:
|
||||
return f"{prefix}-{uuid4().hex[:12]}"
|
||||
Reference in New Issue
Block a user