1196de6e88
Flask backend with Vue 3 frontend for shop floor machine management. Includes database schema export for MySQL shopdb_flask database. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
"""Plugin registry for tracking installed and enabled plugins."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime
|
|
|
|
|
|
@dataclass
|
|
class PluginState:
|
|
"""Persistent state for a plugin."""
|
|
|
|
name: str
|
|
version: str
|
|
installed_at: str
|
|
enabled: bool = True
|
|
migrations_applied: List[str] = field(default_factory=list)
|
|
config: Dict = field(default_factory=dict)
|
|
|
|
|
|
class PluginRegistry:
|
|
"""
|
|
Manages plugin state persistence.
|
|
Stores state in JSON file in instance folder.
|
|
"""
|
|
|
|
def __init__(self, state_file: Path):
|
|
self.state_file = state_file
|
|
self._plugins: Dict[str, PluginState] = {}
|
|
self._load()
|
|
|
|
def _load(self) -> None:
|
|
"""Load registry from file."""
|
|
if self.state_file.exists():
|
|
try:
|
|
with open(self.state_file, 'r') as f:
|
|
data = json.load(f)
|
|
for name, state_data in data.get('plugins', {}).items():
|
|
self._plugins[name] = PluginState(**state_data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
# Corrupted file, start fresh
|
|
self._plugins = {}
|
|
|
|
def _save(self) -> None:
|
|
"""Save registry to file."""
|
|
self.state_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self.state_file, 'w') as f:
|
|
json.dump({
|
|
'plugins': {
|
|
name: asdict(state)
|
|
for name, state in self._plugins.items()
|
|
}
|
|
}, f, indent=2)
|
|
|
|
def register(self, name: str, version: str) -> PluginState:
|
|
"""Register a newly installed plugin."""
|
|
state = PluginState(
|
|
name=name,
|
|
version=version,
|
|
installed_at=datetime.utcnow().isoformat(),
|
|
enabled=True
|
|
)
|
|
self._plugins[name] = state
|
|
self._save()
|
|
return state
|
|
|
|
def unregister(self, name: str) -> None:
|
|
"""Remove plugin from registry."""
|
|
if name in self._plugins:
|
|
del self._plugins[name]
|
|
self._save()
|
|
|
|
def get(self, name: str) -> Optional[PluginState]:
|
|
"""Get plugin state."""
|
|
return self._plugins.get(name)
|
|
|
|
def is_installed(self, name: str) -> bool:
|
|
"""Check if plugin is installed."""
|
|
return name in self._plugins
|
|
|
|
def is_enabled(self, name: str) -> bool:
|
|
"""Check if plugin is enabled."""
|
|
state = self._plugins.get(name)
|
|
return state.enabled if state else False
|
|
|
|
def enable(self, name: str) -> None:
|
|
"""Enable a plugin."""
|
|
if name in self._plugins:
|
|
self._plugins[name].enabled = True
|
|
self._save()
|
|
|
|
def disable(self, name: str) -> None:
|
|
"""Disable a plugin."""
|
|
if name in self._plugins:
|
|
self._plugins[name].enabled = False
|
|
self._save()
|
|
|
|
def get_enabled_plugins(self) -> List[str]:
|
|
"""Get list of enabled plugin names."""
|
|
return [
|
|
name for name, state in self._plugins.items()
|
|
if state.enabled
|
|
]
|
|
|
|
def add_migration(self, name: str, revision: str) -> None:
|
|
"""Record that a migration was applied."""
|
|
if name in self._plugins:
|
|
if revision not in self._plugins[name].migrations_applied:
|
|
self._plugins[name].migrations_applied.append(revision)
|
|
self._save()
|
|
|
|
def get_all(self) -> Dict[str, PluginState]:
|
|
"""Get all registered plugins."""
|
|
return self._plugins.copy()
|
|
|
|
def update_config(self, name: str, config: Dict) -> None:
|
|
"""Update plugin configuration."""
|
|
if name in self._plugins:
|
|
self._plugins[name].config.update(config)
|
|
self._save()
|