Coverage for src/lektor_ng/pluginsystem.py: 99%
135 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-05 10:32 +0000
1from __future__ import annotations
3import inspect
4import os
5import sys
6import warnings
7from importlib import metadata
8from pathlib import Path
9from weakref import ref as weakref
11from lektor_ng.context import get_ctx
12from lektor_ng.inifile import IniFile
13from lektor_ng.utils import process_extra_flags, split_camel_case
16def get_plugin(plugin_id_or_class, env=None):
17 """Looks up the plugin instance by id or class."""
18 if env is None:
19 ctx = get_ctx()
20 if ctx is None:
21 raise RuntimeError("Context is unavailable and no environment was passed to the function.")
22 env = ctx.env
23 plugin_id = env.plugin_ids_by_class.get(plugin_id_or_class, plugin_id_or_class)
24 try:
25 return env.plugins[plugin_id]
26 except KeyError as error:
27 raise LookupError(f"Plugin {plugin_id!r} not found") from error
30class Plugin:
31 """This needs to be subclassed for custom plugins."""
33 name: str
34 description: str
36 __dist: metadata.Distribution | None = None
38 def __init__(self, env, id):
39 self._env = weakref(env)
40 self.id = id
42 @property
43 def name(self) -> str:
44 """Provide a default value for the plugin name.
46 This default name is formed from the plugin class name, which is split on
47 camel-case word boundaries, with any trailing "Plugin" removed.
49 Note that if you don't like this default, you may override it by setting
50 a `description` attribute on your subclass.
52 """
53 words = split_camel_case(self.__class__.__name__)
54 if len(words) > 1 and words[-1].title() == "Plugin":
55 del words[-1]
56 return " ".join(words)
58 @property
59 def description(self) -> str:
60 """Provide a default description from the plugin distribution's metadata.
62 This default is taken from the value for `description` key set in the
63 `[project]` section of `pyproject.toml` (or the `description` parameter passed
64 to `setup()`).
66 Note that if you don't like this default, you may override it by setting
67 a `description` attribute directly on your subclass.
69 """
70 if self.__dist is not None:
71 # The "Summary" is, confusingly, set, e.g. in pyproject.toml's
72 # project.description key.
73 return self.__dist.metadata["Summary"]
74 return "<no description available>"
76 @property
77 def env(self):
78 rv = self._env()
79 if rv is None:
80 raise RuntimeError("Environment went away")
81 return rv
83 @property
84 def version(self):
85 if self.__dist is not None:
86 return self.__dist.version
87 return None
89 @property
90 def path(self) -> str | None:
91 mod = sys.modules[self.__class__.__module__.split(".", maxsplit=1)[0]]
92 if mod.__file__ is None:
93 return None # pragma: no cover
94 path = Path(mod.__file__).resolve().parent
95 if path.is_relative_to(self.env.project.get_package_cache_path()):
96 # We're only interested in local, editable packages. This is not one.
97 return None
98 return os.fspath(path)
100 @property
101 def import_name(self):
102 return self.__class__.__module__ + ":" + self.__class__.__name__
104 def get_lektor_config(self):
105 """Returns the global config."""
106 ctx = get_ctx()
107 if ctx is not None:
108 cfg = ctx.pad.db.config
109 else:
110 cfg = self.env.load_config()
111 return cfg
113 @property
114 def config_filename(self):
115 """The filename of the plugin specific config file."""
116 return os.path.join(self.env.root_path, "configs", self.id + ".ini")
118 def get_config(self, fresh=False):
119 """Returns the config specific for this plugin. By default this
120 will be cached for the current build context but this can be
121 disabled by passing ``fresh=True``.
122 """
123 ctx = get_ctx()
124 if ctx is not None and not fresh:
125 cache = ctx.cache.setdefault(__name__ + ":configs", {})
126 cfg = cache.get(self.id)
127 if cfg is None:
128 cfg = IniFile(self.config_filename)
129 cache[self.id] = cfg
130 else:
131 cfg = IniFile(self.config_filename)
132 if ctx is not None:
133 ctx.record_dependency(self.config_filename)
134 return cfg
136 def emit(self, event, **kwargs):
137 return self.env.plugin_controller.emit(self.id + "-" + event, **kwargs)
139 def to_json(self):
140 return {
141 "id": self.id,
142 "name": self.name,
143 "version": self.version,
144 "description": self.description,
145 "path": self.path,
146 "import_name": self.import_name,
147 }
150def _check_dist_name(dist_name, plugin_id):
151 """Check that plugin comes from a validly named distribution.
153 Raises RuntimeError if distribution name is not of the form
154 ``lektor-``*<plugin_id>*.
155 """
156 # XXX: Do we really need to be so strict about distribution names?
157 # Ref: https://github.com/lektor/lektor/issues/875
158 match_name = "lektor-" + plugin_id.lower()
159 if match_name != dist_name.lower():
160 raise RuntimeError(
161 "Disallowed distribution name: distribution name for "
162 f"plugin {plugin_id!r} must be {match_name!r} (not {dist_name!r})."
163 )
166def initialize_plugins(env):
167 """Initializes the plugins for the environment."""
168 for ep in metadata.entry_points(group="lektor.plugins"):
169 if ep.dist is not None:
170 _check_dist_name(ep.dist.metadata["Name"], ep.name)
171 plugin_id = ep.name
172 plugin_cls = ep.load()
173 env.plugin_controller.instanciate_plugin(plugin_id, plugin_cls, ep.dist)
174 env.plugin_controller.emit("setup-env")
177class PluginController:
178 """Helper management class that is used to control plugins through
179 the environment.
180 """
182 def __init__(self, env, extra_flags=None):
183 self._env = weakref(env)
184 self.extra_flags = extra_flags
186 @property
187 def env(self):
188 rv = self._env()
189 if rv is None:
190 raise RuntimeError("Environment went away")
191 return rv
193 def instanciate_plugin(
194 self,
195 plugin_id: str,
196 plugin_cls: type[Plugin],
197 dist: metadata.Distribution | None = None,
198 ) -> None:
199 env = self.env
200 if plugin_id in env.plugins:
201 raise RuntimeError(f'Plugin "{plugin_id}" is already registered')
202 plugin = plugin_cls(env, plugin_id)
203 # Plugin.version needs the source distribution to be able to cleanly determine
204 # the plugin version. For reasons of backward compatibility, we don't want to
205 # change the signature of the constructor, so we stick it in a private attribute
206 # here.
207 plugin._Plugin__dist = dist
208 env.plugins[plugin_id] = plugin
209 env.plugin_ids_by_class[plugin_cls] = plugin_id
211 def iter_plugins(self):
212 # XXX: sort?
213 return self.env.plugins.values()
215 def emit(self, event, **kwargs):
216 """Invoke event hook for all plugins that support it.
218 Any ``kwargs`` are passed to the hook methods.
220 Returns a dict mapping plugin ids to hook method return values.
221 """
222 rv = {}
223 extra_flags = process_extra_flags(self.extra_flags)
224 funcname = "on_" + event.replace("-", "_")
225 for plugin in self.iter_plugins():
226 handler = getattr(plugin, funcname, None)
227 if handler is not None:
228 kw = {**kwargs, "extra_flags": extra_flags}
229 try:
230 inspect.signature(handler).bind(**kw)
231 except TypeError:
232 del kw["extra_flags"]
233 rv[plugin.id] = handler(**kw)
234 if "extra_flags" not in kw:
235 warnings.warn(
236 # deprecated since 3.2.0
237 f"The plugin {plugin.id!r} function {funcname!r} does not "
238 "accept extra_flags. "
239 "It should be updated to accept `**extra` so that it will "
240 "not break if new parameters are passed to it by newer "
241 "versions of Lektor.",
242 DeprecationWarning,
243 stacklevel=2,
244 )
245 return rv