shopdb-flask/shopdb/plugins/registry.py

"""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, timezone


@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 = {}
        self._migrate_renamed_plugins()

    def _migrate_renamed_plugins(self) -> None:
        """Upgrade path for the equipment -> machines plugin rename.

        Existing installs carry an 'equipment' entry in plugins.json. When the
        renamed plugins/machines directory exists and no 'machines' entry does,
        carry the state over under the new name and persist.
        """
        if 'equipment' not in self._plugins or 'machines' in self._plugins:
            return
        machines_dir = Path(__file__).resolve().parents[2] / 'plugins' / 'machines'
        if not machines_dir.exists():
            return
        state = self._plugins.pop('equipment')
        state.name = 'machines'
        self._plugins['machines'] = state
        self._save()

    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, enabled: bool = True) -> PluginState:
        """Register a newly installed plugin. enabled=False leaves it off until a
        site opts in (used for plugins that provision extra tables)."""
        state = PluginState(
            name=name,
            version=version,
            installed_at=datetime.now(timezone.utc).replace(tzinfo=None).isoformat(),
            enabled=enabled
        )
        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()