Newer
Older
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
import persisting_theory
import django.dispatch
from django import apps
import logging
from . import config
logger = logging.getLogger(__name__)
class Plugin(apps.AppConfig):
_is_funkwhale_plugin = True
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.hooks = HookRegistry()
self.settings = SettingRegistry()
self.user_settings = SettingRegistry()
def ready(self):
super().ready()
logging.info("Loading plugin %s…", self.label)
self.load()
logging.info("Plugin %s loaded", self.label)
def load(self):
pass
class FuncRegistry(persisting_theory.Registry):
def connect(self, hook_name):
def inner(handler):
self[hook_name] = handler
return handler
return inner
class HookRegistry(FuncRegistry):
pass
class SettingRegistry(persisting_theory.Registry):
def prepare_name(self, data, name):
return data().identifier()
class PluginException(Exception):
pass
class PluginNotFound(PluginException):
pass
class Skip(PluginException):
pass
class PluginSignal(object):
def __init__(self, name, providing_args=[]):
self.name = name
self.providing_args = providing_args
class Hook(PluginSignal):
pass
class SignalsRegistry(persisting_theory.Registry):
def prepare_name(self, data, name):
return data.name
def dispatch(self, hook_name, plugins_conf, **kwargs):
"""
Call all handlers connected to hook_name in turn.
"""
if hook_name not in self:
raise LookupError(hook_name)
logger.debug("[Plugin:hook:%s] Dispatching hook", hook_name)
matching_hooks = []
for row in plugins_conf:
try:
matching_hooks.append((row, row["obj"].hooks[hook_name]))
except KeyError:
continue
if matching_hooks:
logger.debug(
"[Plugin:hook:%s] %s handlers found", hook_name, len(matching_hooks)
)
else:
logger.debug("[Plugin:hook:%s] No handler founds", hook_name)
return
for row, handler in matching_hooks:
logger.debug(
"[Plugin:hook:%s] Calling handler %s from plugin %s",
hook_name,
handler,
row["obj"].name,
)
try:
handler(plugin_conf=row, **kwargs)
except Skip:
logger.debug("[Plugin:hook:%s] handler skipped", hook_name)
except Exception:
logger.exception(
"[Plugin:hook:%s] unknown exception with handler %s",
hook_name,
handler,
)
else:
logger.debug("[Plugin:hook:%s] handler %s called successfully", handler)
logger.debug("[Plugin:hook:%s] Done", hook_name)
hooks = SignalsRegistry()
def get_plugin(name):
try:
plugin = apps.apps.get_app_config(name)
except LookupError:
raise PluginNotFound(name)
if not getattr(plugin, "_is_funkwhale_plugin", False):
raise PluginNotFound(name)
return plugin
def get_all_plugins():
return [
app
for app in apps.apps.get_app_configs()
if getattr(app, "_is_funkwhale_plugin", False)
]
def generate_plugins_conf(plugins, user=None):
qs = models.Plugin.objects.filter(is_enabled=True).values("name", "config")
by_plugin_name = {obj["name"]: obj["config"] for obj in qs}
for plugin in plugins:
if plugin.name not in by_plugin_name:
continue
conf = {
"obj": plugin,
"user": None,
"settings": by_plugin_name[plugin.name] or {},
}
if plugins_conf and user and user.is_authenticated:
qs = models.UserPlugin.objects.filter(
user=user, plugin__is_enabled=True, is_enabled=True
).values("plugin__name", "config")
by_plugin_name = {obj["plugin__name"]: obj["config"] for obj in qs}
if row["obj"].name in by_plugin_name:
row["user"] = {
"id": user.pk,
"settings": by_plugin_name[row["obj"].name],
}
def attach_plugins_conf(obj, user):
from funkwhale_api.common import preferences
plugins_enabled = preferences.get("plugins__enabled")
if plugins_enabled:
conf = generate_plugins_conf(plugins=get_all_plugins(), user=user)
setattr(obj, "plugins_conf", conf)