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
from __future__ import annotations
import importlib.metadata
import logging
from collections.abc import Callable, Iterable
from pygitweb.api.action import Action
from pygitweb.api.subpage import Subpage
from pygitweb.config import ACTIONS as BUILTIN_ACTIONS
logger = logging.getLogger(__name__)
SUBPAGE_GROUP = "pygitweb.subpages"
ACTION_GROUP = "pygitweb.actions"
PLUGIN_ACTIONS: dict[str, Action] = {}
SUBPAGES: list[Subpage] = []
DISPATCH_ACTIONS: set[str] = set(BUILTIN_ACTIONS)
def _entry_points(group: str) -> Iterable[importlib.metadata.EntryPoint]:
return importlib.metadata.entry_points().select(group=group)
def _dist_name(ep: importlib.metadata.EntryPoint) -> str | None:
dist = ep.dist
if dist is None:
return None
return dist.name
def _enabled_set(enabled: frozenset[str] | None) -> frozenset[str]:
if enabled is not None:
return enabled
from pygitweb.config import settings
return frozenset(settings.ENABLED_PLUGINS)
def list_available_plugins() -> list[str]:
"""Unique sorted distribution names that declare pygitweb entry points."""
names: set[str] = set()
for group in (SUBPAGE_GROUP, ACTION_GROUP):
for ep in _entry_points(group):
name = _dist_name(ep)
if name:
names.add(name)
return sorted(names)
def _register_plugin(
register: Callable[[type], type],
loaded: object,
ep_name: str,
kind: str,
) -> type | None:
if not isinstance(loaded, type):
logger.error("pygitweb: entry point %r must be a %s class, got %r", ep_name, kind, loaded)
return None
try:
register(loaded)
except TypeError as exc:
logger.error(
"pygitweb: entry point %r must implement %s, got %r (%s)",
ep_name,
kind,
loaded,
exc,
)
return None
return loaded
def load_subpages(enabled: frozenset[str] | None = None) -> list[Subpage]:
allowed = _enabled_set(enabled)
out: list[Subpage] = []
seen_names: set[str] = set()
for ep in _entry_points(SUBPAGE_GROUP):
dist = _dist_name(ep)
if dist is None or dist not in allowed:
continue
try:
loaded = ep.load()
except Exception:
logger.exception("pygitweb: failed to load subpage entry point %r", ep.name)
continue
plugin_cls = _register_plugin(Subpage.register, loaded, ep.name, "Subpage")
if plugin_cls is None:
continue
try:
instance = plugin_cls()
except Exception:
logger.exception("pygitweb: failed to instantiate subpage %r", ep.name)
continue
key = instance.subpage_name()
if key in seen_names:
logger.warning("pygitweb: duplicate subpage %r from %r; skipping", key, ep.name)
continue
seen_names.add(key)
out.append(instance)
return out
def load_plugin_actions(enabled: frozenset[str] | None = None) -> dict[str, Action]:
allowed = _enabled_set(enabled)
out: dict[str, Action] = {}
for ep in _entry_points(ACTION_GROUP):
dist = _dist_name(ep)
if dist is None or dist not in allowed:
continue
try:
loaded = ep.load()
except Exception:
logger.exception("pygitweb: failed to load action entry point %r", ep.name)
continue
plugin_cls = _register_plugin(Action.register, loaded, ep.name, "Action")
if plugin_cls is None:
continue
try:
instance = plugin_cls()
except Exception:
logger.exception("pygitweb: failed to instantiate action %r", ep.name)
continue
name = instance.action_name()
if name in BUILTIN_ACTIONS:
logger.warning(
"pygitweb: plugin action %r conflicts with built-in; skipping entry %r",
name,
ep.name,
)
continue
if name in out:
logger.warning("pygitweb: duplicate plugin action %r; skipping entry %r", name, ep.name)
continue
out[name] = instance
return out
def reload_plugins() -> tuple[dict[str, Action], list[Subpage]]:
"""Re-read ENABLED_PLUGINS and return freshly loaded actions and subpages."""
from pygitweb.config import settings
allowed = frozenset(settings.ENABLED_PLUGINS)
return load_plugin_actions(allowed), load_subpages(allowed)
def apply_plugin_reload() -> None:
"""Reload plugin actions/subpages from ENABLED_PLUGINS into the runtime registry."""
actions, subpages = reload_plugins()
PLUGIN_ACTIONS.clear()
PLUGIN_ACTIONS.update(actions)
SUBPAGES.clear()
SUBPAGES.extend(subpages)
DISPATCH_ACTIONS.clear()
DISPATCH_ACTIONS.update(BUILTIN_ACTIONS)
DISPATCH_ACTIONS.update(PLUGIN_ACTIONS.keys())