Coverage for src/mkdocstrings/plugin.py: 91.50%
140 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-14 19:41 +0100
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-14 19:41 +0100
1"""This module contains the "mkdocstrings" plugin for MkDocs.
3The plugin instantiates a Markdown extension ([`MkdocstringsExtension`][mkdocstrings.extension.MkdocstringsExtension]),
4and adds it to the list of Markdown extensions used by `mkdocs`
5during the [`on_config` event hook](https://www.mkdocs.org/user-guide/plugins/#on_config).
7Once the documentation is built, the [`on_post_build` event hook](https://www.mkdocs.org/user-guide/plugins/#on_post_build)
8is triggered and calls the [`handlers.teardown()` method][mkdocstrings.handlers.base.Handlers.teardown]. This method is
9used to teardown the handlers that were instantiated during documentation buildup.
11Finally, when serving the documentation, it can add directories to watch
12during the [`on_serve` event hook](https://www.mkdocs.org/user-guide/plugins/#on_serve).
13"""
15from __future__ import annotations
17import datetime
18import functools
19import os
20import sys
21from concurrent import futures
22from io import BytesIO
23from typing import TYPE_CHECKING, Any, Callable, Iterable, List, Mapping, Tuple, TypeVar
25from mkdocs.config import Config
26from mkdocs.config import config_options as opt
27from mkdocs.plugins import BasePlugin
28from mkdocs.utils import write_file
29from mkdocs_autorefs.plugin import AutorefsPlugin
31from mkdocstrings._cache import download_and_cache_url, download_url_with_gz
32from mkdocstrings.extension import MkdocstringsExtension
33from mkdocstrings.handlers.base import BaseHandler, Handlers
34from mkdocstrings.loggers import get_logger
36if TYPE_CHECKING:
37 from jinja2.environment import Environment
38 from mkdocs.config.defaults import MkDocsConfig
40if sys.version_info < (3, 10):
41 from typing_extensions import ParamSpec
42else:
43 from typing import ParamSpec
45log = get_logger(__name__)
47InventoryImportType = List[Tuple[str, Mapping[str, Any]]]
48InventoryLoaderType = Callable[..., Iterable[Tuple[str, str]]]
50P = ParamSpec("P")
51R = TypeVar("R")
54def list_to_tuple(function: Callable[P, R]) -> Callable[P, R]:
55 """Decorater to convert lists to tuples in the arguments."""
57 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
58 safe_args = [tuple(item) if isinstance(item, list) else item for item in args]
59 if kwargs: 59 ↛ 61line 59 didn't jump to line 61, because the condition on line 59 was never false
60 kwargs = {key: tuple(value) if isinstance(value, list) else value for key, value in kwargs.items()} # type: ignore[assignment]
61 return function(*safe_args, **kwargs) # type: ignore[arg-type]
63 return wrapper
66class PluginConfig(Config):
67 """The configuration options of `mkdocstrings`, written in `mkdocs.yml`."""
69 handlers = opt.Type(dict, default={})
70 """
71 Global configuration of handlers.
73 You can set global configuration per handler, applied everywhere,
74 but overridable in each "autodoc" instruction. Example:
76 ```yaml
77 plugins:
78 - mkdocstrings:
79 handlers:
80 python:
81 options:
82 option1: true
83 option2: "value"
84 rust:
85 options:
86 option9: 2
87 ```
88 """
90 default_handler = opt.Type(str, default="python")
91 """The default handler to use. The value is the name of the handler module. Default is "python"."""
92 custom_templates = opt.Optional(opt.Dir(exists=True))
93 """Location of custom templates to use when rendering API objects.
95 Value should be the path of a directory relative to the MkDocs configuration file.
96 """
97 enable_inventory = opt.Optional(opt.Type(bool))
98 """Whether to enable object inventory creation."""
99 enabled = opt.Type(bool, default=True)
100 """Whether to enable the plugin. Default is true. If false, *mkdocstrings* will not collect or render anything."""
103class MkdocstringsPlugin(BasePlugin[PluginConfig]):
104 """An `mkdocs` plugin.
106 This plugin defines the following event hooks:
108 - `on_config`
109 - `on_env`
110 - `on_post_build`
112 Check the [Developing Plugins](https://www.mkdocs.org/user-guide/plugins/#developing-plugins) page of `mkdocs`
113 for more information about its plugin system.
114 """
116 css_filename = "assets/_mkdocstrings.css"
118 def __init__(self) -> None:
119 """Initialize the object."""
120 super().__init__()
121 self._handlers: Handlers | None = None
123 @property
124 def handlers(self) -> Handlers:
125 """Get the instance of [mkdocstrings.handlers.base.Handlers][] for this plugin/build.
127 Raises:
128 RuntimeError: If the plugin hasn't been initialized with a config.
130 Returns:
131 An instance of [mkdocstrings.handlers.base.Handlers][] (the same throughout the build).
132 """
133 if not self._handlers: 133 ↛ 134line 133 didn't jump to line 134, because the condition on line 133 was never true
134 raise RuntimeError("The plugin hasn't been initialized with a config yet")
135 return self._handlers
137 def on_config(self, config: MkDocsConfig) -> MkDocsConfig | None:
138 """Instantiate our Markdown extension.
140 Hook for the [`on_config` event](https://www.mkdocs.org/user-guide/plugins/#on_config).
141 In this hook, we instantiate our [`MkdocstringsExtension`][mkdocstrings.extension.MkdocstringsExtension]
142 and add it to the list of Markdown extensions used by `mkdocs`.
144 We pass this plugin's configuration dictionary to the extension when instantiating it (it will need it
145 later when processing markdown to get handlers and their global configurations).
147 Arguments:
148 config: The MkDocs config object.
150 Returns:
151 The modified config.
152 """
153 if not self.plugin_enabled:
154 log.debug("Plugin is not enabled. Skipping.")
155 return config
156 log.debug("Adding extension to the list")
158 theme_name = config.theme.name or os.path.dirname(config.theme.dirs[0])
160 to_import: InventoryImportType = []
161 for handler_name, conf in self.config.handlers.items():
162 for import_item in conf.pop("import", ()):
163 if isinstance(import_item, str): 163 ↛ 165line 163 didn't jump to line 165, because the condition on line 163 was never false
164 import_item = {"url": import_item} # noqa: PLW2901
165 to_import.append((handler_name, import_item))
167 extension_config = {
168 "theme_name": theme_name,
169 "mdx": config.markdown_extensions,
170 "mdx_configs": config.mdx_configs,
171 "mkdocstrings": self.config,
172 "mkdocs": config,
173 }
174 self._handlers = Handlers(extension_config)
176 autorefs: AutorefsPlugin
177 try:
178 # If autorefs plugin is explicitly enabled, just use it.
179 autorefs = config.plugins["autorefs"] # type: ignore[assignment]
180 log.debug(f"Picked up existing autorefs instance {autorefs!r}")
181 except KeyError:
182 # Otherwise, add a limited instance of it that acts only on what's added through `register_anchor`.
183 autorefs = AutorefsPlugin()
184 autorefs.scan_toc = False
185 config.plugins["autorefs"] = autorefs
186 log.debug(f"Added a subdued autorefs instance {autorefs!r}")
187 # Add collector-based fallback in either case.
188 autorefs.get_fallback_anchor = self.handlers.get_anchors
190 mkdocstrings_extension = MkdocstringsExtension(extension_config, self.handlers, autorefs)
191 config.markdown_extensions.append(mkdocstrings_extension) # type: ignore[arg-type]
193 config.extra_css.insert(0, self.css_filename) # So that it has lower priority than user files.
195 self._inv_futures = {}
196 if to_import:
197 inv_loader = futures.ThreadPoolExecutor(4)
198 for handler_name, import_item in to_import:
199 loader = self.get_handler(handler_name).load_inventory
200 future = inv_loader.submit(
201 self._load_inventory, # type: ignore[misc]
202 loader,
203 **import_item,
204 )
205 self._inv_futures[future] = (loader, import_item)
206 inv_loader.shutdown(wait=False)
208 return config
210 @property
211 def inventory_enabled(self) -> bool:
212 """Tell if the inventory is enabled or not.
214 Returns:
215 Whether the inventory is enabled.
216 """
217 inventory_enabled = self.config.enable_inventory
218 if inventory_enabled is None: 218 ↛ 220line 218 didn't jump to line 220, because the condition on line 218 was never false
219 inventory_enabled = any(handler.enable_inventory for handler in self.handlers.seen_handlers) 219 ↛ exitline 219 didn't finish the generator expression on line 219
220 return inventory_enabled
222 @property
223 def plugin_enabled(self) -> bool:
224 """Tell if the plugin is enabled or not.
226 Returns:
227 Whether the plugin is enabled.
228 """
229 return self.config.enabled
231 def on_env(self, env: Environment, config: MkDocsConfig, *args: Any, **kwargs: Any) -> None: # noqa: ARG002
232 """Extra actions that need to happen after all Markdown rendering and before HTML rendering.
234 Hook for the [`on_env` event](https://www.mkdocs.org/user-guide/plugins/#on_env).
236 - Write mkdocstrings' extra files into the site dir.
237 - Gather results from background inventory download tasks.
238 """
239 if not self.plugin_enabled:
240 return
241 if self._handlers: 241 ↛ 250line 241 didn't jump to line 250, because the condition on line 241 was never false
242 css_content = "\n".join(handler.extra_css for handler in self.handlers.seen_handlers)
243 write_file(css_content.encode("utf-8"), os.path.join(config.site_dir, self.css_filename))
245 if self.inventory_enabled: 245 ↛ 250line 245 didn't jump to line 250, because the condition on line 245 was never false
246 log.debug("Creating inventory file objects.inv")
247 inv_contents = self.handlers.inventory.format_sphinx()
248 write_file(inv_contents, os.path.join(config.site_dir, "objects.inv"))
250 if self._inv_futures: 250 ↛ exitline 250 didn't return from function 'on_env', because the condition on line 250 was never false
251 log.debug(f"Waiting for {len(self._inv_futures)} inventory download(s)")
252 futures.wait(self._inv_futures, timeout=30)
253 results = {}
254 # Reversed order so that pages from first futures take precedence:
255 for fut in reversed(list(self._inv_futures)):
256 try:
257 results.update(fut.result())
258 except Exception as error: # noqa: BLE001
259 loader, import_item = self._inv_futures[fut]
260 loader_name = loader.__func__.__qualname__
261 log.error(f"Couldn't load inventory {import_item} through {loader_name}: {error}") # noqa: TRY400
262 for page, identifier in results.items():
263 config.plugins["autorefs"].register_url(page, identifier) # type: ignore[attr-defined]
264 self._inv_futures = {}
266 def on_post_build(
267 self,
268 config: MkDocsConfig, # noqa: ARG002
269 **kwargs: Any, # noqa: ARG002
270 ) -> None:
271 """Teardown the handlers.
273 Hook for the [`on_post_build` event](https://www.mkdocs.org/user-guide/plugins/#on_post_build).
274 This hook is used to teardown all the handlers that were instantiated and cached during documentation buildup.
276 For example, a handler could open a subprocess in the background and keep it open
277 to feed it "autodoc" instructions and get back JSON data. If so, it should then close the subprocess at some point:
278 the proper place to do this is in the handler's `teardown` method, which is indirectly called by this hook.
280 Arguments:
281 config: The MkDocs config object.
282 **kwargs: Additional arguments passed by MkDocs.
283 """
284 if not self.plugin_enabled:
285 return
287 for future in self._inv_futures: 287 ↛ 288line 287 didn't jump to line 288, because the loop on line 287 never started
288 future.cancel()
290 if self._handlers: 290 ↛ exitline 290 didn't return from function 'on_post_build', because the condition on line 290 was never false
291 log.debug("Tearing handlers down")
292 self.handlers.teardown()
294 def get_handler(self, handler_name: str) -> BaseHandler:
295 """Get a handler by its name. See [mkdocstrings.handlers.base.Handlers.get_handler][].
297 Arguments:
298 handler_name: The name of the handler.
300 Returns:
301 An instance of a subclass of [`BaseHandler`][mkdocstrings.handlers.base.BaseHandler].
302 """
303 return self.handlers.get_handler(handler_name)
305 @classmethod
306 # lru_cache does not allow mutable arguments such lists, but that is what we load from YAML config.
307 @list_to_tuple
308 @functools.lru_cache(maxsize=None)
309 def _load_inventory(cls, loader: InventoryLoaderType, url: str, **kwargs: Any) -> Mapping[str, str]:
310 """Download and process inventory files using a handler.
312 Arguments:
313 loader: A function returning a sequence of pairs (identifier, url).
314 url: The URL to download and process.
315 **kwargs: Extra arguments to pass to the loader.
317 Returns:
318 A mapping from identifier to absolute URL.
319 """
320 log.debug(f"Downloading inventory from {url!r}")
321 content = download_and_cache_url(url, download_url_with_gz, datetime.timedelta(days=1))
322 result = dict(loader(BytesIO(content), url=url, **kwargs))
323 log.debug(f"Loaded inventory from {url!r}: {len(result)} items")
324 return result