tessl/pypi-bokeh

Interactive plots and applications in the browser from Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Client and Server

Name: tessl/pypi-bokeh
Author: tessl

Client-server functionality for building interactive web applications with real-time Python callbacks. Enables building standalone Bokeh applications that update dynamically based on user interactions, data changes, and server-side computations.

Capabilities

Application Framework

Core classes for creating Bokeh applications that run on the Bokeh server.

class Application:
    """
    Factory for creating Bokeh documents with handlers.
    
    Applications define how to create documents for each client session,
    enabling multi-user interactive applications.
    """
    def __init__(self, *handlers, **kwargs):
        """
        Parameters:
        - handlers: ApplicationHandler objects that define document creation logic
        - metadata: dict, application metadata
        """

def curdoc():
    """
    Get the current document for server applications.
    
    Returns the Document associated with the current Bokeh server session.
    Essential for server applications to access and modify the document.
    
    Returns:
    Document: Current document instance
    """

Client Session Management

Functions and classes for connecting to and managing Bokeh server sessions from client code.

class ClientSession:
    """
    Client session for connecting to a Bokeh server.
    
    Manages the connection between client code and a running Bokeh server,
    enabling programmatic control of server applications.
    """
    def __init__(self, session_id=None, websocket_url=None, io_loop=None, **kwargs):
        """
        Parameters:
        - session_id: str, specific session identifier
        - websocket_url: str, server websocket URL
        - io_loop: IOLoop, Tornado IOLoop instance
        """
    
    def push(self, document=None):
        """Push document changes to the server."""
    
    def pull(self):
        """Pull document state from the server."""
    
    def close(self, why="closed"):
        """Close the client session."""

def pull_session(session_id=None, url=None, io_loop=None):
    """
    Pull an existing session from a Bokeh server.
    
    Parameters:
    - session_id: str, session identifier to pull
    - url: str, server URL
    - io_loop: IOLoop, Tornado IOLoop instance
    
    Returns:
    ClientSession: Connected client session
    """

def push_session(document, session_id=None, url=None, io_loop=None):
    """
    Push a document to create a new server session.
    
    Parameters:
    - document: Document, document to push to server
    - session_id: str, session identifier
    - url: str, server URL
    - io_loop: IOLoop, Tornado IOLoop instance
    
    Returns:
    ClientSession: Connected client session
    """

def show_session(session_id=None, url=None, browser=None, new=None):
    """
    Display a server session in the browser.
    
    Parameters:
    - session_id: str, session identifier
    - url: str, server URL
    - browser: str, browser name to use
    - new: int, browser window/tab behavior
    """

DEFAULT_SESSION_ID: str  # Default session identifier

Server Components

Core server infrastructure for running Bokeh server applications.

class Server:
    """
    Bokeh server for hosting interactive applications.
    
    Runs Bokeh applications and serves them to web browsers,
    handling multiple concurrent sessions and real-time updates.
    """
    def __init__(self, applications=None, port=5006, address=None, **kwargs):
        """
        Parameters:
        - applications: dict, mapping of URL paths to Application objects
        - port: int, server port number
        - address: str, server bind address
        - allow_websocket_origin: list, allowed websocket origins
        - io_loop: IOLoop, Tornado IOLoop instance
        """
    
    def start(self):
        """Start the Bokeh server."""
    
    def stop(self):
        """Stop the Bokeh server."""
    
    def io_loop(self):
        """Get the server's IOLoop."""

class BaseServer:
    """Base class for Bokeh server implementations."""
    
class BokehTornado:
    """Tornado-based Bokeh server implementation."""

Document Lifecycle

Functions for managing document lifecycle in server applications.

def without_document_lock(func):
    """
    Decorator to execute function without document lock.
    
    Used in server applications when modifying documents
    from background threads or async callbacks.
    
    Parameters:
    - func: callable, function to execute without lock
    
    Returns:
    Decorated function
    """

class DocumentLock:
    """Document locking mechanism for thread safety."""
    
    def __enter__(self):
        """Acquire document lock."""
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Release document lock."""

Session State Management

Tools for managing application state across server sessions.

class SessionContext:
    """Context information for server sessions."""
    
    def __init__(self, server_context=None, session_id=None, **kwargs):
        """
        Parameters:
        - server_context: ServerContext, server context
        - session_id: str, session identifier
        """
    
    @property
    def request(self):
        """Get the current HTTP request."""
    
    @property
    def session_id(self):
        """Get the session identifier."""

class ServerContext:
    """Server-level context information."""
    
    def __init__(self, **kwargs):
        """
        Parameters:
        - sessions: dict, active sessions
        - applications: dict, loaded applications
        """

Periodic Callbacks

Functions for scheduling periodic updates in server applications.

def add_periodic_callback(callback, period_milliseconds):
    """
    Schedule a function to be called periodically.
    
    Parameters:
    - callback: callable, function to call periodically
    - period_milliseconds: int, callback interval in milliseconds
    
    Returns:
    Callback handle for removing the periodic callback
    """

def remove_periodic_callback(callback_handle):
    """
    Remove a periodic callback.
    
    Parameters:
    - callback_handle: handle returned by add_periodic_callback
    """

class PeriodicCallback:
    """Periodic callback wrapper."""
    
    def __init__(self, callback, period, io_loop=None):
        """
        Parameters:
        - callback: callable, function to call
        - period: int, callback period in milliseconds
        - io_loop: IOLoop, Tornado IOLoop instance
        """
    
    def start(self):
        """Start the periodic callback."""
    
    def stop(self):
        """Stop the periodic callback."""

Application Handlers

Handler classes for different types of Bokeh applications.

class ApplicationHandler:
    """Base class for application handlers."""
    
    def modify_document(self, doc):
        """Modify document when creating new sessions."""

class FunctionHandler(ApplicationHandler):
    """Handler that wraps a simple function."""
    
    def __init__(self, func):
        """
        Parameters:
        - func: callable, function that modifies a document
        """

class ScriptHandler(ApplicationHandler):
    """Handler that executes a Python script."""
    
    def __init__(self, filename=None, source=None, **kwargs):
        """
        Parameters:
        - filename: str, path to Python script file
        - source: str, Python source code
        """

class DirectoryHandler(ApplicationHandler):
    """Handler for directory-based applications."""
    
    def __init__(self, filename, **kwargs):
        """
        Parameters:
        - filename: str, path to application directory
        """

class NotebookHandler(ApplicationHandler):
    """Handler for Jupyter notebook applications."""
    
    def __init__(self, filename, **kwargs):
        """
        Parameters:
        - filename: str, path to notebook file
        """

Usage Examples

Basic Server Application

from bokeh.application import Application
from bokeh.application.handlers import FunctionHandler
from bokeh.server.server import Server
from bokeh.plotting import figure
from bokeh.models.widgets import Button
from bokeh.layouts import column
from bokeh.io import curdoc

def create_document(doc):
    # Create plot
    plot = figure(width=400, height=400)
    plot.circle([1, 2, 3], [4, 5, 6], size=10)
    
    # Create button
    button = Button(label="Update Plot")
    
    def button_callback():
        # Update plot data
        plot.circle([2, 3, 4], [5, 6, 7], size=15, color="red")
    
    button.on_click(button_callback)
    
    # Add to document
    layout = column(button, plot)
    doc.add_root(layout)

# Create application
app = Application(FunctionHandler(create_document))

# Start server
server = Server({'/': app}, port=5006)
server.start()

print("Server running on http://localhost:5006")
server.io_loop.add_callback(server.show, "/")
server.io_loop.start()

Client Session Management

from bokeh.client import push_session, pull_session
from bokeh.plotting import figure
from bokeh.io import curdoc

# Create a document
plot = figure()
plot.line([1, 2, 3], [4, 5, 6])

# Push to server
session = push_session(curdoc(), url="http://localhost:5006")

# Later, pull from server
session = pull_session(url="http://localhost:5006", session_id="specific_session")
print(f"Document title: {session.document.title}")

# Close session
session.close()

Periodic Updates

from bokeh.io import curdoc
from bokeh.plotting import figure
from bokeh.models import ColumnDataSource
import numpy as np

# Create plot with data source
source = ColumnDataSource(data=dict(x=[1, 2, 3], y=[4, 5, 6]))
plot = figure()
plot.circle('x', 'y', source=source, size=10)

def update_data():
    # Generate new random data
    new_data = dict(
        x=np.random.random(10),
        y=np.random.random(10)
    )
    source.data = new_data

# Add periodic callback (update every 1000ms)
curdoc().add_periodic_callback(update_data, 1000)
curdoc().add_root(plot)

Multi-User Application State

from bokeh.io import curdoc
from bokeh.models.widgets import TextInput, Div
from bokeh.layouts import column

# Shared application state
app_state = {"message_count": 0}

def create_session():
    # Create per-session widgets
    text_input = TextInput(title="Enter message:")
    message_div = Div(text=f"Messages sent: {app_state['message_count']}")
    
    def send_message(attr, old, new):
        # Update shared state
        app_state["message_count"] += 1
        message_div.text = f"Messages sent: {app_state['message_count']}"
        text_input.value = ""
    
    text_input.on_change('value', send_message)
    
    layout = column(text_input, message_div)
    curdoc().add_root(layout)

create_session()

Common Data Types

# Server-related types
ServerContextLike = Union[ServerContext, None]
SessionContextLike = Union[SessionContext, None]
ApplicationLike = Union[Application, str, callable]

# Handler types
HandlerLike = Union[ApplicationHandler, str, callable]
ModifyDocumentCallable = Callable[[Document], None]

# Session types
SessionIdLike = Union[str, None]
URLLike = Union[str, None]
IOLoopLike = Union[Any, None]  # Tornado IOLoop type

Install with Tessl CLI