54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
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]}"
|