Sphinx

# Source code for sphinx.events

    """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[..., Any]
        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, Set[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[..., Any], 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[Any]:
            """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