You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
50 lines
1.7 KiB
50 lines
1.7 KiB
from __future__ import annotations |
|
|
|
import json |
|
import threading |
|
from copy import deepcopy |
|
from pathlib import Path |
|
from typing import Any, Dict |
|
|
|
|
|
class ConfigStore: |
|
def __init__(self, conf_path: Path): |
|
self.conf_path = conf_path |
|
self._lock = threading.RLock() |
|
|
|
def load(self) -> Dict[str, Any]: |
|
with self._lock: |
|
with self.conf_path.open("r", encoding="utf-8") as conf_file: |
|
return json.load(conf_file) |
|
|
|
def save(self, config: Dict[str, Any]) -> None: |
|
with self._lock: |
|
with self.conf_path.open("w", encoding="utf-8") as conf_file: |
|
json.dump(config, conf_file, indent=2) |
|
conf_file.write("\n") |
|
|
|
def get_copy(self) -> Dict[str, Any]: |
|
return deepcopy(self.load()) |
|
|
|
def upsert_trigger(self, trigger_id: str, trigger_payload: Dict[str, Any]) -> Dict[str, Any]: |
|
config = self.load() |
|
config.setdefault("triggers", {})[trigger_id] = trigger_payload |
|
self.save(config) |
|
return config["triggers"][trigger_id] |
|
|
|
def patch_trigger(self, trigger_id: str, trigger_payload: Dict[str, Any]) -> Dict[str, Any]: |
|
config = self.load() |
|
triggers = config.setdefault("triggers", {}) |
|
if trigger_id not in triggers: |
|
raise KeyError(trigger_id) |
|
triggers[trigger_id].update(trigger_payload) |
|
self.save(config) |
|
return triggers[trigger_id] |
|
|
|
def delete_trigger(self, trigger_id: str) -> None: |
|
config = self.load() |
|
triggers = config.setdefault("triggers", {}) |
|
if trigger_id not in triggers: |
|
raise KeyError(trigger_id) |
|
del triggers[trigger_id] |
|
self.save(config)
|
|
|