Source code for bokeh.application.handlers.server_lifecycle

''' Bokeh Application Handler to look for Bokeh server lifecycle callbacks
in a specified Python module.

'''
from __future__ import absolute_import, print_function

import codecs
import os

from .handler import Handler
from .code_runner import CodeRunner

from bokeh.util.callback_manager import _check_callback

def _do_nothing(ignored):
    pass

[docs]class ServerLifecycleHandler(Handler):
    ''' Load a script which contains server lifecycle callbacks.

    '''

[docs]    def __init__(self, *args, **kwargs):
        '''

        Keyword Args:
            filename (str) : path to a module to load lifecycle callbacks from

            argv (list[str], optional) : a list of string arguments to use as
                ``sys.argv`` when the callback code is executed. (default: [])

        '''
        super(ServerLifecycleHandler, self).__init__(*args, **kwargs)

        if 'filename' not in kwargs:
            raise ValueError('Must pass a filename to ServerLifecycleHandler')
        filename = kwargs['filename']
        argv = kwargs.get('argv', [])

        source = codecs.open(filename, 'r', 'UTF-8').read()

        self._runner = CodeRunner(source, filename, argv)

        self._on_server_loaded = _do_nothing
        self._on_server_unloaded = _do_nothing
        self._on_session_created = _do_nothing
        self._on_session_destroyed = _do_nothing

        if not self._runner.failed:
            # unlike ScriptHandler, we only load the module one time
            self._module = self._runner.new_module()

            def extract_callbacks():
                contents = self._module.__dict__
                if 'on_server_loaded' in contents:
                    self._on_server_loaded = contents['on_server_loaded']
                if 'on_server_unloaded' in contents:
                    self._on_server_unloaded = contents['on_server_unloaded']
                if 'on_session_created' in contents:
                    self._on_session_created = contents['on_session_created']
                if 'on_session_destroyed' in contents:
                    self._on_session_destroyed = contents['on_session_destroyed']

                _check_callback(self._on_server_loaded, ('server_context',), what="on_server_loaded")
                _check_callback(self._on_server_unloaded, ('server_context',), what="on_server_unloaded")
                _check_callback(self._on_session_created, ('session_context',), what="on_session_created")
                _check_callback(self._on_session_destroyed, ('session_context',), what="on_session_destroyed")

            self._runner.run(self._module, extract_callbacks)

[docs]    def url_path(self):
        ''' The last path component for the basename of the path to the
        callback module.

        '''
        if self.failed:
            return None
        else:
            # TODO should fix invalid URL characters
            return '/' + os.path.splitext(os.path.basename(self._runner.path))[0]

[docs]    def on_server_loaded(self, server_context):
        ''' Execute `on_server_unloaded`` from the configured module (if
        it is defined) when the server is first started.

        Args:
            server_context (ServerContext) :

        '''
        return self._on_server_loaded(server_context)

[docs]    def on_server_unloaded(self, server_context):
        ''' Execute ``on_server_unloaded`` from the configured module (if
        it is defined) when the server cleanly exits. (Before stopping the
        server's ``IOLoop``.)

        Args:
            server_context (ServerContext) :

        .. warning::
            In practice this code may not run, since servers are often killed
            by a signal.

        '''
        return self._on_server_unloaded(server_context)

[docs]    def on_session_created(self, session_context):
        ''' Execute ``on_session_created`` from the configured module (if
        it is defined) when a new session is created.

        Args:
            session_context (SessionContext) :

        '''
        return self._on_session_created(session_context)

[docs]    def on_session_destroyed(self, session_context):
        ''' Execute ``on_session_destroyed`` from the configured module (if
        it is defined) when a new session is destroyed.

        Args:
            session_context (SessionContext) :

        '''
        return self._on_session_destroyed(session_context)

[docs]    def modify_document(self, doc):
        ''' This handler does not make any modifications to the Document.

        Args:
            doc (Document) : A Bokeh Document to update in-place

                *This handler does not modify the document*

        Returns:
            None

        '''
        # we could support a modify_document function, might be weird though.
        pass

    @property
    def failed(self):
        ''' ``True`` if the lifecycle callbacks failed to execute

        '''
        return self._runner.failed

    @property
    def error(self):
        ''' If the handler fails, may contain a related error message.

        '''
        return self._runner.error

    @property
    def error_detail(self):
        ''' If the handler fails, may contain a traceback or other details.

        '''
        return self._runner.error_detail