mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-10-31 04:01:59 +01:00
feat(plugins): add runtime endpoint and functionality to retrieve plugin runtime data
This commit is contained in:
1 parent
6f9ef1c394
commit
6a393c5c61
8 files changed
+123
-3
No files matched your search
@@ -74,6 +74,7 @@ class PluginIPCServer:
|
||||
"uninstall": self._op_uninstall,
|
||||
"ui_info": self._op_ui_info,
|
||||
"get_config": self._op_get_config,
|
||||
"get_runtime": self._op_get_runtime,
|
||||
"set_config": self._op_set_config,
|
||||
"catalogue": self._op_catalogue,
|
||||
"catalogue_install": self._op_catalogue_install,
|
||||
@@ -244,6 +245,9 @@ class PluginIPCServer:
|
||||
def _op_get_config(self, request: dict[str, Any]) -> dict[str, Any]:
|
||||
return self.manager.get_config(self._require_id(request))
|
||||
|
||||
def _op_get_runtime(self, request: dict[str, Any]) -> dict[str, Any]:
|
||||
return self.manager.get_runtime(self._require_id(request))
|
||||
|
||||
def _op_set_config(self, request: dict[str, Any]) -> dict[str, Any]:
|
||||
config = request.get("config")
|
||||
if not isinstance(config, dict):
|
||||
@@ -364,6 +368,9 @@ class PluginIPCClient:
|
||||
def get_config(self, plugin_id: str) -> dict[str, Any]:
|
||||
return self.call("get_config", plugin_id=plugin_id)
|
||||
|
||||
def get_runtime(self, plugin_id: str) -> dict[str, Any]:
|
||||
return self.call("get_runtime", plugin_id=plugin_id)
|
||||
|
||||
def set_config(
|
||||
self, plugin_id: str, config: dict[str, Any], *, restart: bool = False
|
||||
) -> dict[str, Any]:
|
||||
|
||||
@@ -4,13 +4,12 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import shutil
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
import shutil
|
||||
|
||||
from .catalogue import CatalogueClient, CatalogueError, DEFAULT_CATALOGUE_URL
|
||||
from .catalogue import DEFAULT_CATALOGUE_URL, CatalogueClient, CatalogueError
|
||||
from .github_releases import (
|
||||
GitHubReleaseClient,
|
||||
GitHubReleasesError,
|
||||
@@ -268,6 +267,25 @@ class PluginManager:
|
||||
result["restarted"] = restarted
|
||||
return result
|
||||
|
||||
def get_runtime(self, plugin_id: str) -> dict[str, Any]:
|
||||
"""Return plugin runtime snapshot from data/runtime.json for UIs."""
|
||||
state = self.storage.read_state(plugin_id)
|
||||
if state is None:
|
||||
raise PluginManagerError(f"plugin not found: {plugin_id}", 404)
|
||||
paths = self.storage.paths_for(plugin_id)
|
||||
paths.data_dir.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
runtime = self.storage.read_runtime(plugin_id)
|
||||
except ValueError as exc:
|
||||
raise PluginManagerError(str(exc), 500) from exc
|
||||
path = self.storage.runtime_path(plugin_id)
|
||||
return {
|
||||
"id": plugin_id,
|
||||
"path": str(path),
|
||||
"exists": path.is_file(),
|
||||
"runtime": runtime,
|
||||
}
|
||||
|
||||
def uninstall(self, plugin_id: str, *, delete_data: bool = False) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
state = self.storage.read_state(plugin_id)
|
||||
|
||||
@@ -225,6 +225,10 @@ class PluginStorage:
|
||||
"""Path to the plugin-owned config.json under data/."""
|
||||
return self.paths_for(plugin_id).data_dir / "config.json"
|
||||
|
||||
def runtime_path(self, plugin_id: str) -> Path:
|
||||
"""Path to the plugin runtime snapshot under data/."""
|
||||
return self.paths_for(plugin_id).data_dir / "runtime.json"
|
||||
|
||||
def read_config(self, plugin_id: str) -> dict[str, Any]:
|
||||
"""Read plugin-owned data/config.json. Missing file → {}."""
|
||||
path = self.config_path(plugin_id)
|
||||
@@ -250,6 +254,21 @@ class PluginStorage:
|
||||
self._atomic_write_json(path, config)
|
||||
return path
|
||||
|
||||
def read_runtime(self, plugin_id: str) -> dict[str, Any]:
|
||||
"""Read plugin-owned data/runtime.json. Missing file -> {}."""
|
||||
path = self.runtime_path(plugin_id)
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"invalid plugin runtime.json: {exc}") from exc
|
||||
if data is None:
|
||||
return {}
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("plugin runtime.json must be a JSON object")
|
||||
return data
|
||||
|
||||
def tail_log(self, plugin_id: str, lines: int = 200) -> list[str]:
|
||||
paths = self.paths_for(plugin_id)
|
||||
log_path = paths.log_file
|
||||
|
||||
Reference in new issue
Block a user