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 = {}

    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()