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,
)

# 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]


[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("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



[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
#-----------------------------------------------------------------------------