"""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.deprecation import _deprecation_warning
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._event_listeners import (
_AutodocBeforeProcessSignatureListener,
_AutodocProcessBasesListener,
_AutodocProcessDocstringListener,
_AutodocProcessSignatureListener,
_AutodocSkipMemberListener,
)
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',
}
[docs]
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
# pass through errors for debugging.
self._reraise_errors: bool = app.pdb
[docs]
def add(self, name: str) -> None:
"""Register a custom Sphinx event called *name*.
This is needed to be able to emit the event.
:param name: The name of the event.
"""
if name in self.events:
msg = __('Event %r already present')
raise ExtensionError(msg % name)
self.events[name] = ''
@property
def app(self) -> Sphinx:
_deprecation_warning(__name__, 'EventManager.app', remove=(11, 0))
return self._app
# ---- 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: _AutodocBeforeProcessSignatureListener,
priority: int,
) -> int: ...
@overload
def connect(
self,
name: Literal['autodoc-process-signature'],
callback: _AutodocProcessSignatureListener,
priority: int,
) -> int: ...
@overload
def connect(
self,
name: Literal['autodoc-process-bases'],
callback: _AutodocProcessBasesListener,
priority: int,
) -> int: ...
@overload
def connect(
self,
name: Literal['autodoc-skip-member'],
callback: _AutodocSkipMemberListener,
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: ...
[docs]
def connect(self, name: str, callback: Callable[..., Any], priority: int) -> int:
"""Connect a handler to specific event.
Register *callback* to be called when the *name* event is emitted.
See :ref:`event callbacks <events>` for details on available core events
and the arguments of their corresponding callback functions.
:param name:
The name of the target event.
:param callback:
Callback function for the event.
:param priority:
The priority of the callback.
The callbacks will be invoked in ascending order of *priority*.
:return:
A listener ID, for use with the :meth:`disconnect` method.
.. versionchanged:: 3.0
Support *priority*
"""
if name not in self.events:
msg = __('Unknown event name: %s')
raise ExtensionError(msg % name)
listener_id = self.next_listener_id
self.next_listener_id += 1
self.listeners[name].append(EventListener(listener_id, callback, priority))
return listener_id
[docs]
def disconnect(self, listener_id: int) -> None:
"""Disconnect the handler given by *listener_id*.
:param listener_id:
A listener_id previously returned by :meth:`connect`.
"""
for listeners in self.listeners.values():
for listener in listeners.copy():
if listener.id == listener_id:
listeners.remove(listener)
[docs]
def emit(
self,
name: str,
*args: Any,
allowed_exceptions: tuple[type[Exception], ...] = (),
) -> list[Any]:
"""Emit a Sphinx event.
This emits the *name* event and passes *args* to the handler functions.
Return the return values of all handlers as a list.
Do not emit core Sphinx events in extensions!
:param name:
The name of the event that will be emitted.
:param args:
The arguments for the event, to be passed to the handler functions.
:param allowed_exceptions:
The list of exceptions that are allowed in the handlers.
.. versionchanged:: 3.1
Added *allowed_exceptions* to specify path-through exceptions
"""
# 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._reraise_errors:
raise
modname = safe_getattr(listener.handler, '__module__', None)
msg = __('Handler %r for event %r threw an exception')
raise ExtensionError(
msg % (listener.handler, name),
exc,
modname=modname,
) from exc
return results
[docs]
def emit_firstresult(
self,
name: str,
*args: Any,
allowed_exceptions: tuple[type[Exception], ...] = (),
) -> Any:
"""Emit a Sphinx event and return the first result.
This emits the *name* event and passes *args* to the handler functions.
The first non-None result is returned.
:param name:
The name of the event that will be emitted.
:param args:
The arguments for the event, to be passed to the handler functions.
:param allowed_exceptions:
The list of exceptions that are allowed in the handlers.
.. versionadded:: 0.5
.. versionchanged:: 3.1
Added *allowed_exceptions* to specify path-through exceptions
"""
for result in self.emit(name, *args, allowed_exceptions=allowed_exceptions):
if result is not None:
return result
return None
