/
plugin_utils.py
174 lines (154 loc) · 6.8 KB
/
plugin_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
import importlib.util
import os
import sys
from importlib.abc import Loader
from typing import Any, Dict
import sentry_sdk
from flask import Flask, Blueprint
def register_plugins(app: Flask):
"""
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:
# this is deprecated behaviour which we should remove in version 1.0
app.logger.debug(
"No plugins configured. Attempting deprecated setting FLEXMEASURES_PLUGIN_PATHS ..."
)
plugins = app.config.get("FLEXMEASURES_PLUGIN_PATHS", [])
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)
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",
},
}
"""
assert isinstance(settings, dict), f"{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], 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: Any
):
"""Log a message for this config setting that has the wrong type."""
app.logger.warning(
f"Config setting '{setting_name}' is a {type(setting)} 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"
getattr(app.logger, level)(
f"Missing config setting '{setting_name}'{description}.{message_if_missing}",
)