Source code for bokeh.application.handlers.handler
#-----------------------------------------------------------------------------
# Copyright (c) 2012 - 2022, 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, Dict
# 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
#-----------------------------------------------------------------------------
[docs]class Handler:
''' Provide a mechanism for Bokeh applications to build up new Bokeh
Documents.
.. autoclasstoc::
'''
_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 ----------------------------------------------------------
[docs] 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()")
[docs] 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
[docs] 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
[docs] 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
[docs] 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
[docs] 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 {}
[docs] 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
[docs] 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
#-----------------------------------------------------------------------------