1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
"""
sphinx.events
~~~~~~~~~~~~~
Sphinx core events.
Gracefully adapted from the TextPress system by Armin.
:copyright: Copyright 2007-2020 by the Sphinx team, see AUTHORS.
:license: BSD, see LICENSE for details.
"""
import warnings
from collections import OrderedDict, defaultdict
from typing import Any, Callable, Dict, List
from sphinx.deprecation import RemovedInSphinx40Warning
from sphinx.errors import ExtensionError
from sphinx.locale import __
from sphinx.util import logging
if False:
# For type annotation
from sphinx.application import Sphinx
logger = logging.getLogger(__name__)
# List of all known core events. Maps name to arguments description.
core_events = {
'builder-inited': '',
'config-inited': 'config',
'env-get-outdated': 'env, added, changed, removed',
'env-get-updated': 'env',
'env-purge-doc': 'env, docname',
'env-before-read-docs': 'env, docnames',
'env-check-consistency': 'env',
'source-read': 'docname, source text',
'doctree-read': 'the doctree before being pickled',
'env-merge-info': 'env, read docnames, other env instance',
'missing-reference': 'env, node, contnode',
'doctree-resolved': 'doctree, docname',
'env-updated': 'env',
'html-collect-pages': 'builder',
'html-page-context': 'pagename, context, doctree or None',
'build-finished': 'exception',
}
class EventManager:
"""Event manager for Sphinx."""
def __init__(self, app: "Sphinx" = None) -> None:
if app is None:
warnings.warn('app argument is required for EventManager.',
RemovedInSphinx40Warning)
self.app = app
self.events = core_events.copy()
self.listeners = defaultdict(OrderedDict) # type: Dict[str, Dict[int, Callable]]
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] = ''
def connect(self, name: str, callback: Callable) -> 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][listener_id] = callback
return listener_id
def disconnect(self, listener_id: int) -> None:
"""Disconnect a handler."""
for event in self.listeners.values():
event.pop(listener_id, None)
def emit(self, name: str, *args: Any) -> List:
"""Emit a Sphinx event."""
try:
logger.debug('[app] emitting event: %r%s', name, repr(args)[:100])
except Exception:
# not every object likes to be repr()'d (think
# random stuff coming via autodoc)
pass
results = []
for callback in self.listeners[name].values():
if self.app is None:
# for compatibility; RemovedInSphinx40Warning
results.append(callback(*args))
else:
results.append(callback(self.app, *args))
return results
def emit_firstresult(self, name: str, *args: Any) -> 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):
if result is not None:
return result
return None
|
