""" Utils for registering FlexMeasures plugins """ from __future__ import annotations import importlib.util import os import sys from importlib.abc import Loader from types import ModuleType import sentry_sdk from flask import Flask, Blueprint from flexmeasures.utils.coding_utils import get_classes_module def register_plugins(app: Flask): # noqa: C901 """ Register FlexMeasures plugins as Blueprints. This is configured by the config setting FLEXMEASURES_PLUGINS. Assumptions: - a setting EITHER points to a plugin folder containing an __init__.py file OR it is the name of an installed module, which can be imported. - each plugin defines at least one Blueprint object. These will be registered with the Flask app, so their functionality (e.g. routes) becomes available. If you load a plugin via a file path, we'll refer to the plugin with the name of your plugin folder (last part of the path). """ plugins = app.config.get("FLEXMEASURES_PLUGINS", []) if not plugins and "FLEXMEASURES_PLUGIN_PATHS" in app.config: app.logger.warning( "Plugins found via FLEXMEASURES_PLUGIN_PATHS. This setting will be sunset in v0.14. Please switch to FLEXMEASURES_PLUGINS." ) plugins = app.config.get("FLEXMEASURES_PLUGIN_PATHS", []) if isinstance(plugins, str): plugins = [ plugin.strip() for plugin in plugins.split(",") if len(plugin.strip()) > 0 ] if not isinstance(plugins, list): app.logger.error( f"The value of FLEXMEASURES_PLUGINS is not a list: {plugins}. Cannot install plugins ..." ) return app.config["LOADED_PLUGINS"] = {} for plugin in plugins: plugin_name = plugin.split("/")[-1] app.logger.info(f"Importing plugin {plugin_name} ...") module = None if not os.path.exists(plugin): # assume plugin is a package pkg_name = os.path.split(plugin)[ -1 ] # rule out attempts for relative package imports app.logger.debug( f"Attempting to import {pkg_name} as an installed package ..." ) try: module = importlib.import_module(pkg_name) except ModuleNotFoundError: app.logger.error( f"Attempted to import module {pkg_name} (as it is not a valid file path), but it is not installed." ) continue else: # assume plugin is a file path if not os.path.exists(os.path.join(plugin, "__init__.py")): app.logger.error( f"Plugin {plugin_name} is a valid file path, but does not contain an '__init__.py' file. Cannot load plugin {plugin_name}." ) continue spec = importlib.util.spec_from_file_location( plugin_name, os.path.join(plugin, "__init__.py") ) if spec is None: app.logger.error( f"Could not load specs for plugin {plugin_name} at {plugin}." ) continue module = importlib.util.module_from_spec(spec) sys.modules[plugin_name] = module assert isinstance(spec.loader, Loader) spec.loader.exec_module(module) if module is None: app.logger.error(f"Plugin {plugin} could not be loaded.") continue plugin_version = getattr(module, "__version__", "0.1") plugin_settings = getattr(module, "__settings__", {}) check_config_settings(app, plugin_settings) # Look for blueprints in the plugin's main __init__ module and register them plugin_blueprints = [ getattr(module, a) for a in dir(module) if isinstance(getattr(module, a), Blueprint) ] if not plugin_blueprints: app.logger.warning( f"No blueprints found for plugin {plugin_name} at {plugin}." ) continue for plugin_blueprint in plugin_blueprints: app.logger.debug(f"Registering {plugin_blueprint} ...") app.register_blueprint(plugin_blueprint) # Load reporters and schedulers from flexmeasures.data.models.reporting import Reporter from flexmeasures.data.models.planning import Scheduler plugin_reporters = get_classes_module(module.__name__, Reporter) plugin_schedulers = get_classes_module(module.__name__, Scheduler) # add DataGenerators if plugin_reporters: app.data_generators["reporter"].update(plugin_reporters) if plugin_schedulers: app.data_generators["scheduler"].update(plugin_schedulers) app.config["LOADED_PLUGINS"][plugin_name] = plugin_version app.logger.info(f"Loaded plugins: {app.config['LOADED_PLUGINS']}") sentry_sdk.set_context("plugins", app.config.get("LOADED_PLUGINS", {})) def check_config_settings(app, settings: dict[str, dict]): """Make sure expected config settings exist. For example: settings = { "MY_PLUGIN_URL": { "description": "URL used by my plugin for x.", "level": "error", }, "MY_PLUGIN_TOKEN": { "description": "Token used by my plugin for y.", "level": "warning", "message": "Without this token, my plugin will not do y.", "parse_as": str, }, "MY_PLUGIN_COLOR": { "description": "Color used to override the default plugin color.", "level": "info", }, } """ # Check config settings are in dict form, after possibly converting them from module variables if isinstance(settings, ModuleType): settings = { setting: settings.__dict__[setting] for setting in dir(settings) if not setting.startswith("__") } assert isinstance(settings, dict), f"{type(settings)} should be a dict" for setting_name, setting_fields in settings.items(): assert isinstance(setting_fields, dict), f"{setting_name} should be a dict" missing_config_settings = [] config_settings_with_wrong_type = [] for setting_name, setting_fields in settings.items(): setting = app.config.get(setting_name) if setting is None: missing_config_settings.append(setting_name) elif "parse_as" in setting_fields and not isinstance( setting, setting_fields["parse_as"] ): config_settings_with_wrong_type.append((setting_name, setting)) for setting_name, setting in config_settings_with_wrong_type: log_wrong_type_for_config_setting( app, setting_name, settings[setting_name], type(setting) ) for setting_name in missing_config_settings: log_missing_config_setting(app, setting_name, settings[setting_name]) def log_wrong_type_for_config_setting( app, setting_name: str, setting_fields: dict, setting_type: type ): """Log a message for this config setting that has the wrong type.""" app.logger.warning( f"Config setting '{setting_name}' is a {setting_type} whereas a {setting_fields['parse_as']} was expected." ) def log_missing_config_setting(app, setting_name: str, setting_fields: dict): """Log a message for this missing config setting. The logging level is taken from the 'level' key. If missing, we default to error. If present, we also log the 'description' and the 'message_if_missing' keys. """ message_if_missing = ( f" {setting_fields['message_if_missing']}" if "message_if_missing" in setting_fields else "" ) description = ( f" ({setting_fields['description']})" if "description" in setting_fields else "" ) level = setting_fields["level"] if "level" in setting_fields else "error" if not hasattr(app.logger, level): app.logger.warning( f"Unrecognized logger level '{level}' for config setting '{setting_name}'." ) level = "error" getattr(app.logger, level)( f"Missing config setting '{setting_name}'{description}.{message_if_missing}", )