-
-
Notifications
You must be signed in to change notification settings - Fork 4.2k
/
handler.py
252 lines (181 loc) · 7.79 KB
/
handler.py
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
#-----------------------------------------------------------------------------
# Copyright (c) 2012 - 2024, Anaconda, Inc., and Bokeh Contributors.
# All rights reserved.
#
# The full license is in the file LICENSE.txt, distributed with this software.
#-----------------------------------------------------------------------------
''' Provide a base class for Bokeh Application handler classes.
When a Bokeh server session is initiated, the Bokeh server asks the Application
for a new Document to service the session. To do this, the Application first
creates a new empty Document, then it passes this new Document to the
``modify_document`` method of each of its handlers. When all handlers have
updated the Document, it is used to service the user session.
Below is an example outline of a custom handler that might modify documents
based off information in some database:
.. code-block:: python
class DatabaseHandler(Handler):
""" A Bokeh Application handler to initialize Documents from a database
"""
def modify_document(self, doc: Document) -> None:
# do some data base lookup here to generate 'plot'
# add the plot to the document (i.e modify the document)
doc.add_root(plot)
'''
#-----------------------------------------------------------------------------
# Boilerplate
#-----------------------------------------------------------------------------
from __future__ import annotations
import logging # isort:skip
log = logging.getLogger(__name__)
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Standard library imports
import os
import sys
import traceback
from typing import TYPE_CHECKING, Any
# Bokeh imports
from ...document import Document
from ..application import ServerContext, SessionContext
if TYPE_CHECKING:
from tornado.httputil import HTTPServerRequest
from .code_runner import CodeRunner
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
__all__ = (
'Handler',
)
#-----------------------------------------------------------------------------
# General API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
class Handler:
''' Provide a mechanism for Bokeh applications to build up new Bokeh
Documents.
'''
_failed: bool
_error: str | None
_error_detail: str | None
_static: str | None
def __init__(self) -> None:
self._failed = False
self._error = None
self._error_detail = None
self._static = None
# Properties --------------------------------------------------------------
@property
def error(self) -> str | None:
''' If the handler fails, may contain a related error message.
'''
return self._error
@property
def error_detail(self) -> str | None:
''' If the handler fails, may contain a traceback or other details.
'''
return self._error_detail
@property
def failed(self) -> bool:
''' ``True`` if the handler failed to modify the doc
'''
return self._failed
@property
def safe_to_fork(self) -> bool:
return True
# Public methods ----------------------------------------------------------
def modify_document(self, doc: Document) -> None:
''' Modify an application document in a specified manner.
When a Bokeh server session is initiated, the Bokeh server asks the
Application for a new Document to service the session. To do this,
the Application first creates a new empty Document, then it passes
this Document to the ``modify_document`` method of each of its
handlers. When all handlers have updated the Document, it is used to
service the user session.
*Subclasses must implement this method*
Args:
doc (Document) : A Bokeh Document to update in-place
Returns:
Document
'''
raise NotImplementedError("implement modify_document()")
def on_server_loaded(self, server_context: ServerContext) -> None:
''' Execute code when the server is first started.
Subclasses may implement this method to provide for any one-time
initialization that is necessary after the server starts, but
before any sessions are created.
Args:
server_context (ServerContext) :
'''
pass
def on_server_unloaded(self, server_context: ServerContext) -> None:
''' Execute code when the server cleanly exits. (Before stopping the
server's ``IOLoop``.)
Subclasses may implement this method to provide for any one-time
tear down that is necessary before the server exits.
Args:
server_context (ServerContext) :
.. warning::
In practice this code may not run, since servers are often killed
by a signal.
'''
pass
async def on_session_created(self, session_context: SessionContext) -> None:
''' Execute code when a new session is created.
Subclasses may implement this method to provide for any per-session
initialization that is necessary before ``modify_doc`` is called for
the session.
Args:
session_context (SessionContext) :
'''
pass
async def on_session_destroyed(self, session_context: SessionContext) -> None:
''' Execute code when a session is destroyed.
Subclasses may implement this method to provide for any per-session
tear-down that is necessary when sessions are destroyed.
Args:
session_context (SessionContext) :
'''
pass
def process_request(self, request: HTTPServerRequest) -> dict[str, Any]:
''' Processes incoming HTTP request returning a dictionary of
additional data to add to the session_context.
Args:
request: HTTP request
Returns:
A dictionary of JSON serializable data to be included on
the session context.
'''
return {}
def static_path(self) -> str | None:
''' Return a path to app-specific static resources, if applicable.
'''
if self.failed:
return None
else:
return self._static
def url_path(self) -> str | None:
''' Returns a default URL path, if applicable.
Handlers subclasses may optionally implement this method, to inform
the Bokeh application what URL it should be installed at.
If multiple handlers specify ``url_path`` the Application will use the
value from the first handler in its list of handlers.
'''
return None
def handle_exception(handler: Handler | CodeRunner, e: Exception) -> None:
''' Record an exception and details on a Handler.
'''
handler._failed = True
handler._error_detail = traceback.format_exc()
_, _, exc_traceback = sys.exc_info()
filename, line_number, func, txt = traceback.extract_tb(exc_traceback)[-1]
basename = os.path.basename(filename)
handler._error = f"{e}\nFile {basename!r}, line {line_number}, in {func}:\n{txt}"
#-----------------------------------------------------------------------------
# Private API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------