Source code for bokeh.application.application

#-----------------------------------------------------------------------------
# Copyright (c) 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,
    TypeAlias,
)

# Bokeh imports
from ..core.types import ID
from ..document import Document
from ..settings import settings

if TYPE_CHECKING:
    from tornado.httputil import HTTPServerRequest

    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]

[docs] 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
[docs] 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 ----------------------------------------------------------
[docs] 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(f"More than one static path requested for app: {list(static_paths)!r}") elif len(static_paths) == 1: self._static_path = static_paths.pop() else: self._static_path = None
[docs] def create_document(self) -> Document: ''' Creates and initializes a document using the Application's handlers. ''' doc = Document() self.initialize_document(doc) return doc
[docs] 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()
[docs] 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)
[docs] 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)
[docs] 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
[docs] 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
[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. ''' request_data: dict[str, Any] = {} for h in self._handlers: request_data.update(h.process_request(request)) return request_data
[docs] 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
[docs] 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
[docs] 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 ----------------------------------------------------------
[docs] @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 #-----------------------------------------------------------------------------