-
-
Notifications
You must be signed in to change notification settings - Fork 4.2k
/
application.py
358 lines (263 loc) · 11.3 KB
/
application.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
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
#-----------------------------------------------------------------------------
# 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 the ``Application`` class.
Application instances are factories for creating new Bokeh Documents.
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.
'''
#-----------------------------------------------------------------------------
# Boilerplate
#-----------------------------------------------------------------------------
from __future__ import annotations
import logging # isort:skip
log = logging.getLogger(__name__)
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Standard library imports
from abc import ABCMeta, abstractmethod
from typing import (
TYPE_CHECKING,
Any,
Awaitable,
Callable,
ClassVar,
)
# Bokeh imports
from ..core.types import ID
from ..document import Document
from ..settings import settings
if TYPE_CHECKING:
from tornado.httputil import HTTPServerRequest
from typing_extensions import TypeAlias
from ..server.session import ServerSession
from .handlers.handler import Handler
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
__all__ = (
'Application',
'ServerContext',
'SessionContext',
)
#-----------------------------------------------------------------------------
# General API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
Callback: TypeAlias = Callable[[], None]
class Application:
''' An Application is a factory for Document instances.
'''
# This is so that bokeh.io.show can check if a passed in object is an
# Application without having to import Application directly. This module
# depends on tornado and we have made a commitment that "basic" modules
# will function without bringing in tornado.
_is_a_bokeh_application_class: ClassVar[bool] = True
_static_path: str | None
_handlers: list[Handler]
_metadata: dict[str, Any] | None
def __init__(self, *handlers: Handler, metadata: dict[str, Any] | None = None) -> None:
''' Application factory.
Args:
handlers (seq[Handler]): List of handlers to call.
The URL is taken from the first one only.
Keyword Args:
metadata (dict): arbitrary user-supplied JSON data to make available
with the application.
The server will provide a URL ``http://applicationurl/metadata``
which returns a JSON blob of the form:
.. code-block:: json
{
"data": {
"hi": "hi",
"there": "there"
},
"url": "/myapp"
}
The user-supplied metadata is returned as-is under the
``"data"`` key in the blob.
'''
self._static_path = None
self._handlers = []
self._metadata = metadata
for h in handlers:
self.add(h)
# Properties --------------------------------------------------------------
@property
def handlers(self) -> tuple[Handler, ...]:
''' The ordered list of handlers this Application is configured with.
'''
return tuple(self._handlers)
@property
def metadata(self) -> dict[str, Any] | None:
''' Arbitrary user-supplied metadata to associate with this application.
'''
return self._metadata
@property
def safe_to_fork(self) -> bool:
'''
'''
return all(handler.safe_to_fork for handler in self._handlers)
@property
def static_path(self) -> str | None:
''' Path to any (optional) static resources specified by handlers.
'''
return self._static_path
# Public methods ----------------------------------------------------------
def add(self, handler: Handler) -> None:
''' Add a handler to the pipeline used to initialize new documents.
Args:
handler (Handler) : a handler for this Application to use to
process Documents
'''
self._handlers.append(handler)
# make sure there is at most one static path
static_paths = {h.static_path() for h in self.handlers}
static_paths.discard(None)
if len(static_paths) > 1:
raise RuntimeError("More than one static path requested for app: %r" % list(static_paths))
elif len(static_paths) == 1:
self._static_path = static_paths.pop()
else:
self._static_path = None
def create_document(self) -> Document:
''' Creates and initializes a document using the Application's handlers.
'''
doc = Document()
self.initialize_document(doc)
return doc
def initialize_document(self, doc: Document) -> None:
''' Fills in a new document using the Application's handlers.
'''
for h in self._handlers:
# TODO (havocp) we need to check the 'failed' flag on each handler
# and build a composite error display. In develop mode, we want to
# somehow get these errors to the client.
h.modify_document(doc)
if h.failed:
log.error("Error running application handler %r: %s %s ", h, h.error, h.error_detail)
if settings.perform_document_validation():
doc.validate()
def on_server_loaded(self, server_context: ServerContext) -> None:
''' Invoked to execute code when a new session is created.
This method calls ``on_server_loaded`` on each handler, in order,
with the server context passed as the only argument.
'''
for h in self._handlers:
h.on_server_loaded(server_context)
def on_server_unloaded(self, server_context: ServerContext) -> None:
''' Invoked to execute code when the server cleanly exits. (Before
stopping the server's ``IOLoop``.)
This method calls ``on_server_unloaded`` on each handler, in order,
with the server context passed as the only argument.
.. warning::
In practice this code may not run, since servers are often killed
by a signal.
'''
for h in self._handlers:
h.on_server_unloaded(server_context)
async def on_session_created(self, session_context: SessionContext) -> None:
''' Invoked to execute code when a new session is created.
This method calls ``on_session_created`` on each handler, in order,
with the session context passed as the only argument.
May return a ``Future`` which will delay session creation until the
``Future`` completes.
'''
for h in self._handlers:
await h.on_session_created(session_context)
return None
async def on_session_destroyed(self, session_context: SessionContext) -> None:
''' Invoked to execute code when a session is destroyed.
This method calls ``on_session_destroyed`` on each handler, in order,
with the session context passed as the only argument.
Afterwards, ``session_context.destroyed`` will be ``True``.
'''
for h in self._handlers:
await h.on_session_destroyed(session_context)
return None
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.
'''
request_data: dict[str, Any] = {}
for h in self._handlers:
request_data.update(h.process_request(request))
return request_data
class ServerContext(metaclass=ABCMeta):
''' A harness for server-specific information and tasks related to
collections of Bokeh sessions.
*This base class is probably not of interest to general users.*
'''
# Properties --------------------------------------------------------------
@property
@abstractmethod
def sessions(self) -> list[ServerSession]:
''' ``SessionContext`` instances belonging to this application.
*Subclasses must implement this method.*
'''
pass
class SessionContext(metaclass=ABCMeta):
''' A harness for server-specific information and tasks related to
Bokeh sessions.
*This base class is probably not of interest to general users.*
'''
_server_context: ServerContext
_id: ID
def __init__(self, server_context: ServerContext, session_id: ID) -> None:
'''
'''
self._server_context = server_context
self._id = session_id
# Properties --------------------------------------------------------------
@property
@abstractmethod
def destroyed(self) -> bool:
''' If ``True``, the session has been discarded and cannot be used.
A new session with the same ID could be created later but this instance
will not come back to life.
'''
pass
@property
def id(self) -> ID:
''' The unique ID for the session associated with this context.
'''
return self._id
@property
def server_context(self) -> ServerContext:
''' The server context for this session context
'''
return self._server_context
# Public methods ----------------------------------------------------------
@abstractmethod
def with_locked_document(self, func: Callable[[Document], Awaitable[None]]) -> Awaitable[None]:
''' Runs a function with the document lock held, passing the
document to the function.
*Subclasses must implement this method.*
Args:
func (callable): function that takes a single parameter (the Document)
and returns ``None`` or a ``Future``
Returns:
a ``Future`` containing the result of the function
'''
pass
SessionDestroyedCallback = Callable[[SessionContext], None]
#-----------------------------------------------------------------------------
# Private API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------