"""Sphinx core events. Gracefully adapted from the TextPress system by Armin. """ from __future__ import annotations from collections import defaultdict from operator import attrgetter from typing import TYPE_CHECKING, NamedTuple, overload from sphinx.errors import ExtensionError, SphinxError from sphinx.locale import __ from sphinx.util import logging from sphinx.util.inspect import safe_getattr if TYPE_CHECKING: from collections.abc import Callable, Iterable, Sequence, Set from pathlib import Path from typing import Any, Literal from docutils import nodes from sphinx import addnodes from sphinx.application import Sphinx from sphinx.builders import Builder from sphinx.config import Config from sphinx.domains import Domain from sphinx.environment import BuildEnvironment from sphinx.ext.autodoc import _AutodocProcessDocstringListener from sphinx.ext.todo import todo_node logger = logging.getLogger(__name__) class EventListener(NamedTuple): id: int handler: Callable priority: int # List of all known core events. Maps name to arguments description. core_events = { 'config-inited': 'config', 'builder-inited': '', 'env-get-outdated': 'env, added, changed, removed', 'env-before-read-docs': 'env, docnames', 'env-purge-doc': 'env, docname', 'source-read': 'docname, source text', 'include-read': 'relative path, parent docname, source text', 'doctree-read': 'the doctree before being pickled', 'env-merge-info': 'env, read docnames, other env instance', 'env-updated': 'env', 'env-get-updated': 'env', 'env-check-consistency': 'env', 'write-started': 'builder', 'doctree-resolved': 'doctree, docname', 'missing-reference': 'env, node, contnode', 'warn-missing-reference': 'domain, node', 'build-finished': 'exception', } class EventManager: """Event manager for Sphinx.""" def __init__(self, app: Sphinx) -> None: self.app = app self.events = core_events.copy() self.listeners: dict[str, list[EventListener]] = defaultdict(list) self.next_listener_id = 0 def add(self, name: str) -> None: """Register a custom Sphinx event.""" if name in self.events: raise ExtensionError(__('Event %r already present') % name) self.events[name] = '' # ---- Core events ------------------------------------------------------- @overload def connect( self, name: Literal['config-inited'], callback: Callable[[Sphinx, Config], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['builder-inited'], callback: Callable[[Sphinx], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-get-outdated'], callback: Callable[ [Sphinx, BuildEnvironment, Set[str], Set[str], Set[str]], Sequence[str] ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-before-read-docs'], callback: Callable[[Sphinx, BuildEnvironment, list[str]], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-purge-doc'], callback: Callable[[Sphinx, BuildEnvironment, str], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['source-read'], callback: Callable[[Sphinx, str, list[str]], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['include-read'], callback: Callable[[Sphinx, Path, str, list[str]], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['doctree-read'], callback: Callable[[Sphinx, nodes.document], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-merge-info'], callback: Callable[ [Sphinx, BuildEnvironment, list[str], BuildEnvironment], None ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-updated'], callback: Callable[[Sphinx, BuildEnvironment], str], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-get-updated'], callback: Callable[[Sphinx, BuildEnvironment], Iterable[str]], priority: int, ) -> int: ... @overload def connect( self, name: Literal['env-check-consistency'], callback: Callable[[Sphinx, BuildEnvironment], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['write-started'], callback: Callable[[Sphinx, Builder], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['doctree-resolved'], callback: Callable[[Sphinx, nodes.document, str], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['missing-reference'], callback: Callable[ [Sphinx, BuildEnvironment, addnodes.pending_xref, nodes.TextElement], nodes.reference | None, ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['warn-missing-reference'], callback: Callable[[Sphinx, Domain, addnodes.pending_xref], bool | None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['build-finished'], callback: Callable[[Sphinx, Exception | None], None], priority: int, ) -> int: ... # ---- Events from builtin builders -------------------------------------- @overload def connect( self, name: Literal['html-collect-pages'], callback: Callable[[Sphinx], Iterable[tuple[str, dict[str, Any], str]]], priority: int, ) -> int: ... @overload def connect( self, name: Literal['html-page-context'], callback: Callable[ [Sphinx, str, str, dict[str, Any], nodes.document], str | None ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['linkcheck-process-uri'], callback: Callable[[Sphinx, str], str | None], priority: int, ) -> int: ... # ---- Events from builtin extensions-- ---------------------------------- @overload def connect( self, name: Literal['object-description-transform'], callback: Callable[[Sphinx, str, str, addnodes.desc_content], None], priority: int, ) -> int: ... # ---- Events from first-party extensions -------------------------------- @overload def connect( self, name: Literal['autodoc-process-docstring'], callback: _AutodocProcessDocstringListener, priority: int, ) -> int: ... @overload def connect( self, name: Literal['autodoc-before-process-signature'], callback: Callable[[Sphinx, Any, bool], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['autodoc-process-signature'], callback: Callable[ [ Sphinx, Literal[ 'module', 'class', 'exception', 'function', 'method', 'attribute' ], str, Any, dict[str, bool], str | None, str | None, ], tuple[str | None, str | None] | None, ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['autodoc-process-bases'], callback: Callable[[Sphinx, str, Any, dict[str, bool], list[str]], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['autodoc-skip-member'], callback: Callable[ [ Sphinx, Literal[ 'module', 'class', 'exception', 'function', 'method', 'attribute' ], str, Any, bool, dict[str, bool], ], bool, ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['todo-defined'], callback: Callable[[Sphinx, todo_node], None], priority: int, ) -> int: ... @overload def connect( self, name: Literal['viewcode-find-source'], callback: Callable[ [Sphinx, str], tuple[str, dict[str, tuple[Literal['class', 'def', 'other'], int, int]]], ], priority: int, ) -> int: ... @overload def connect( self, name: Literal['viewcode-follow-imported'], callback: Callable[[Sphinx, str, str], str | None], priority: int, ) -> int: ... # ---- Catch-all --------------------------------------------------------- @overload def connect( self, name: str, callback: Callable[..., Any], priority: int, ) -> int: ... def connect(self, name: str, callback: Callable, priority: int) -> int: """Connect a handler to specific event.""" if name not in self.events: raise ExtensionError(__('Unknown event name: %s') % name) listener_id = self.next_listener_id self.next_listener_id += 1 self.listeners[name].append(EventListener(listener_id, callback, priority)) return listener_id def disconnect(self, listener_id: int) -> None: """Disconnect a handler.""" for listeners in self.listeners.values(): for listener in listeners.copy(): if listener.id == listener_id: listeners.remove(listener) def emit( self, name: str, *args: Any, allowed_exceptions: tuple[type[Exception], ...] = (), ) -> list: """Emit a Sphinx event.""" # not every object likes to be repr()'d (think # random stuff coming via autodoc) try: repr_args = repr(args) except Exception: pass else: logger.debug('[app] emitting event: %r%s', name, repr_args) results = [] listeners = sorted(self.listeners[name], key=attrgetter('priority')) for listener in listeners: try: results.append(listener.handler(self.app, *args)) except allowed_exceptions: # pass through the errors specified as *allowed_exceptions* raise except SphinxError: raise except Exception as exc: if self.app.pdb: # Just pass through the error, so that it can be debugged. raise modname = safe_getattr(listener.handler, '__module__', None) raise ExtensionError( __('Handler %r for event %r threw an exception') % (listener.handler, name), exc, modname=modname, ) from exc return results def emit_firstresult( self, name: str, *args: Any, allowed_exceptions: tuple[type[Exception], ...] = (), ) -> Any: """Emit a Sphinx event and returns first result. This returns the result of the first handler that doesn't return ``None``. """ for result in self.emit(name, *args, allowed_exceptions=allowed_exceptions): if result is not None: return result return None