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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
import typing as _t
from importlib import metadata as _importlib_metadata
import typing_extensions as _tx
import flask as _f
from .exceptions import HookError
from .resources import ResourceType
from ._callback import ClientsideFuncType
if _t.TYPE_CHECKING:
from .dash import Dash
from .development.base_component import Component
ComponentType = _t.TypeVar("ComponentType", bound=Component)
LayoutType = _t.Union[ComponentType, _t.List[ComponentType]]
else:
LayoutType = None
ComponentType = None
Dash = None
HookDataType = _tx.TypeVar("HookDataType")
# pylint: disable=too-few-public-methods
class _Hook(_tx.Generic[HookDataType]):
def __init__(self, func, priority=0, final=False, data: HookDataType = None):
self.func = func
self.final = final
self.data = data
self.priority = priority
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
class _Hooks:
def __init__(self) -> None:
self._ns = {
"setup": [],
"layout": [],
"routes": [],
"error": [],
"callback": [],
"index": [],
"custom_data": [],
}
self._js_dist = []
self._css_dist = []
self._clientside_callbacks: _t.List[
_t.Tuple[ClientsideFuncType, _t.Any, _t.Any]
] = []
# final hooks are a single hook added to the end of regular hooks.
self._finals = {}
def add_hook(
self,
hook: str,
func: _t.Callable,
priority: _t.Optional[int] = None,
final: bool = False,
data: _t.Optional[HookDataType] = None,
):
if final:
existing = self._finals.get(hook)
if existing:
raise HookError("Final hook already present")
self._finals[hook] = _Hook(func, final, data=data)
return
hks = self._ns.get(hook, [])
p = priority or 0
if not priority and len(hks):
# Take the minimum value and remove 1 to keep the order.
priority_min = min(h.priority for h in hks)
p = priority_min - 1
hks.append(_Hook(func, priority=p, data=data))
self._ns[hook] = sorted(hks, reverse=True, key=lambda h: h.priority)
def get_hooks(self, hook: str) -> _t.List[_Hook]:
final = self._finals.get(hook, None)
if final:
final = [final]
else:
final = []
return self._ns.get(hook, []) + final
def layout(self, priority: _t.Optional[int] = None, final: bool = False):
"""
Run a function when serving the layout, the return value
will be used as the layout.
"""
def _wrap(func: _t.Callable[[LayoutType], LayoutType]):
self.add_hook("layout", func, priority=priority, final=final)
return func
return _wrap
def setup(self, priority: _t.Optional[int] = None, final: bool = False):
"""
Can be used to get a reference to the app after it is instantiated.
"""
def _setup(func: _t.Callable[[Dash], None]):
self.add_hook("setup", func, priority=priority, final=final)
return func
return _setup
def route(
self,
name: _t.Optional[str] = None,
methods: _t.Sequence[str] = ("GET",),
priority: _t.Optional[int] = None,
final: bool = False,
):
"""
Add a route to the Dash server.
"""
def wrap(func: _t.Callable[[], _f.Response]):
_name = name or func.__name__
self.add_hook(
"routes",
func,
priority=priority,
final=final,
data=dict(name=_name, methods=methods),
)
return func
return wrap
def error(self, priority: _t.Optional[int] = None, final: bool = False):
"""Automatically add an error handler to the dash app."""
def _error(func: _t.Callable[[Exception], _t.Any]):
self.add_hook("error", func, priority=priority, final=final)
return func
return _error
def callback(
self, *args, priority: _t.Optional[int] = None, final: bool = False, **kwargs
):
"""
Add a callback to all the apps with the hook installed.
"""
def wrap(func):
self.add_hook(
"callback",
func,
priority=priority,
final=final,
data=(list(args), dict(kwargs)),
)
return func
return wrap
def clientside_callback(
self, clientside_function: ClientsideFuncType, *args, **kwargs
):
"""
Add a callback to all the apps with the hook installed.
"""
self._clientside_callbacks.append((clientside_function, args, kwargs))
def script(self, distribution: _t.List[ResourceType]):
"""Add js scripts to the page."""
self._js_dist.extend(distribution)
def stylesheet(self, distribution: _t.List[ResourceType]):
"""Add stylesheets to the page."""
self._css_dist.extend(distribution)
def index(self, priority: _t.Optional[int] = None, final=False):
"""Modify the index of the apps."""
def wrap(func):
self.add_hook(
"index",
func,
priority=priority,
final=final,
)
return func
return wrap
def custom_data(
self, namespace: str, priority: _t.Optional[int] = None, final=False
):
"""
Add data to the callback_context.custom_data property under the namespace.
The hook function takes the current context_value and before the ctx is set
and has access to the flask request context.
"""
def wrap(func: _t.Callable[[_t.Dict], _t.Any]):
self.add_hook(
"custom_data",
func,
priority=priority,
final=final,
data={"namespace": namespace},
)
return func
return wrap
hooks = _Hooks()
class HooksManager:
# Flag to only run `register_setuptools` once
_registered = False
hooks = hooks
# pylint: disable=too-few-public-methods
class HookErrorHandler:
def __init__(self, original):
self.original = original
def __call__(self, err: Exception):
result = None
if self.original:
result = self.original(err)
hook_result = None
for hook in HooksManager.get_hooks("error"):
hook_result = hook(err)
return result or hook_result
@classmethod
def get_hooks(cls, hook: str):
return cls.hooks.get_hooks(hook)
@classmethod
def register_setuptools(cls):
if cls._registered:
# Only have to register once.
return
for dist in _importlib_metadata.distributions():
for entry in dist.entry_points:
# Look for setup.py entry points named `dash_hooks`
if entry.group != "dash_hooks":
continue
entry.load()
|
