129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
"""File-locked JSON registry for local state management.
|
|
|
|
Provides a simple JSON-backed key-value store with ``fcntl`` file
|
|
locking for safe concurrent access. Useful for CLI tools that need
|
|
to track remote resources (runners, VMs, deployments) on the local
|
|
machine.
|
|
|
|
Usage::
|
|
|
|
from devx.utils.json_registry import JsonRegistry
|
|
|
|
registry = JsonRegistry(Path("~/.local/share/myapp/state.json"))
|
|
registry.add("item1", host="10.0.0.1", user="deploy")
|
|
info = registry.get("item1")
|
|
registry.remove("item1")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import fcntl
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
|
|
class JsonRegistry:
|
|
"""Manages a local JSON file mapping names to arbitrary metadata.
|
|
|
|
Uses ``fcntl`` for file locking (shared lock for reads, exclusive
|
|
lock for writes) to prevent race conditions in concurrent scenarios.
|
|
"""
|
|
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
"""Initialise the registry.
|
|
|
|
Args:
|
|
path: Path to the JSON file. Defaults to
|
|
``~/.local/share/devx/registry.json``.
|
|
"""
|
|
self._path = path or Path.home() / ".local" / "share" / "devx" / "registry.json"
|
|
self._data: dict[str, dict[str, Any]] = self._load()
|
|
|
|
def _load(self) -> dict[str, dict[str, Any]]:
|
|
if not self._path.exists():
|
|
return {}
|
|
try:
|
|
with open(self._path) as f:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_SH)
|
|
try:
|
|
data: Any = json.load(f)
|
|
if isinstance(data, dict):
|
|
return cast(dict[str, dict[str, Any]], data)
|
|
finally:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
except (json.JSONDecodeError, OSError):
|
|
pass
|
|
return {}
|
|
|
|
def _save(self) -> None:
|
|
self._path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self._path, "w") as f:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
json.dump(self._data, f, indent=2)
|
|
finally:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
|
|
def add(self, name: str, **fields: Any) -> None:
|
|
"""Register or overwrite an entry in the registry.
|
|
|
|
Args:
|
|
name: Unique key for the entry.
|
|
**fields: Arbitrary metadata fields to store.
|
|
"""
|
|
self._data[name] = {
|
|
**fields,
|
|
"created_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
self._save()
|
|
|
|
def get(self, name: str) -> dict[str, Any] | None:
|
|
"""Retrieve entry metadata by name.
|
|
|
|
Args:
|
|
name: Key to look up.
|
|
|
|
Returns:
|
|
A copy of the entry's metadata, or None if not found.
|
|
"""
|
|
info = self._data.get(name)
|
|
if info:
|
|
return copy.deepcopy(info)
|
|
return None
|
|
|
|
def remove(self, name: str) -> None:
|
|
"""Remove an entry from the registry.
|
|
|
|
Args:
|
|
name: Key to remove. No-op if not found.
|
|
"""
|
|
if name in self._data:
|
|
del self._data[name]
|
|
self._save()
|
|
|
|
def list(self) -> dict[str, dict[str, Any]]:
|
|
"""Return a copy of all registered entries.
|
|
|
|
Returns:
|
|
Dict mapping names to metadata copies.
|
|
"""
|
|
return {name: copy.deepcopy(info) for name, info in self._data.items()}
|
|
|
|
def update(self, name: str, **fields: Any) -> None:
|
|
"""Update fields for an existing entry.
|
|
|
|
Args:
|
|
name: Key to update.
|
|
**fields: Fields to update (None values are skipped).
|
|
|
|
Raises:
|
|
KeyError: If the entry doesn't exist.
|
|
"""
|
|
if name not in self._data:
|
|
raise KeyError(name)
|
|
self._data[name].update({k: v for k, v in fields.items() if v is not None})
|
|
self._save()
|