plugin.py

# app/plugins/example/plugin.py
"""Flask-AAS Plugin API v1 adapter for the reference plugin."""

from pathlib import Path
from typing import Any

from sqlalchemy import inspect

from app.core.extensions import db
from app.plugins import ApplicationPlugin, PluginConfiguration, load_plugin_manifest
from app.plugins.navigation import register_plugin_navigation

from . import models as example_models  # noqa: F401 - register model metadata
from .cli import cli as example_cli
from .models import ExampleSettings, get_example_settings
from .routes import example_bp


class ExamplePlugin(ApplicationPlugin):
    """Small but complete reference implementation of a Flask-AAS plugin."""

    manifest = load_plugin_manifest(Path(__file__).with_name("plugin.toml"))
    plugin_id = manifest.plugin_id
    name = manifest.name
    version = manifest.version
    api_version = manifest.api_version

    def validate_config(self) -> PluginConfiguration:
        """Require a settings row and plugin-managed secret before normal access."""

        endpoint = "example.settings"
        if not inspect(db.engine).has_table(ExampleSettings.__tablename__):
            return PluginConfiguration(
                configured=False,
                reason="Example plugin database schema is not installed.",
                admin_endpoint=endpoint,
            )

        settings = get_example_settings()
        if settings is None:
            return PluginConfiguration(
                configured=False,
                reason="Example plugin configuration has not been completed.",
                admin_endpoint=endpoint,
            )
        if not settings.heading or not settings.heading.strip():
            return PluginConfiguration(
                configured=False,
                reason="Example public heading must not be empty.",
                admin_endpoint=endpoint,
            )
        if not settings.managed_secret:
            return PluginConfiguration(
                configured=False,
                reason="Example managed secret is missing.",
                admin_endpoint=endpoint,
            )

        return PluginConfiguration(configured=True, admin_endpoint=endpoint)

    def clear_secrets(self) -> None:
        """Clear only the secret owned by this plugin on disable."""

        if not inspect(db.engine).has_table(ExampleSettings.__tablename__):
            return
        settings = get_example_settings()
        if settings is not None:
            settings.managed_secret = None

    def get_cli(self):
        """Expose plugin-owned commands through the generic host dispatcher."""

        return example_cli

    def register(self, app: Any) -> None:
        """Register the plugin's Flask surfaces and ordinary navigation link."""

        app.register_blueprint(example_bp)
        navigation_label = self.manifest.navigation_label or self.name
        register_plugin_navigation(
            app,
            plugin_id=self.plugin_id,
            label=navigation_label,
            endpoint="example.index",
        )

        @app.context_processor
        def example_plugin_template_context():
            return {
                "example_navigation_label": navigation_label,
            }


plugin = ExamplePlugin()