Server and Applications

class Application:
    """
    Bokeh application factory for creating documents.
    
    An Application object is a factory for Document instances. It
    defines the structure and behavior of server applications.
    """
    def __init__(self, *handlers, **kwargs):
        """
        Parameters:
        - handlers: FunctionHandler, DirectoryHandler, or ScriptHandler objects
        - **kwargs: Additional application configuration
        """
    
    # Application lifecycle methods
    def create_document(self):
        """Create a new document instance for a session."""
    
    def process_request(self, request):
        """Process incoming HTTP request."""
    
    def on_server_loaded(self, server_context):
        """Called when application is loaded on server."""
    
    def on_server_unloaded(self, server_context):
        """Called when application is unloaded from server."""
    
    def on_session_created(self, session_context):
        """Called when new session is created."""
    
    def on_session_destroyed(self, session_context):
        """Called when session is destroyed."""

# Application handlers
class Handler:
    """Base class for application handlers."""

class FunctionHandler(Handler):
    """Handler that uses a function to create document content."""
    def __init__(self, func):
        """
        Parameters:
        - func: Function that takes a Document and modifies it
        """

class ScriptHandler(Handler):
    """Handler that loads Python script as application."""
    def __init__(self, filename):
        """
        Parameters:
        - filename: Path to Python script file
        """

class DirectoryHandler(Handler):
    """Handler that loads application from directory structure."""
    def __init__(self, path):
        """
        Parameters:
        - path: Path to application directory
        """

def curdoc():
    """
    Get current document.
    
    In server applications, returns the Document instance for the
    current session. In standalone scripts, creates a new Document.
    
    Returns:
    Document: Current document instance
    """

class Document:
    """
    Container for all Bokeh models in an application.
    
    The Document holds all models and manages their relationships,
    callbacks, and lifecycle in server applications.
    """
    def __init__(self, **kwargs): ...
    
    # Model management
    def add_root(self, model):
        """Add a model as a root of the document."""
    
    def remove_root(self, model):
        """Remove a root model from the document."""
    
    @property
    def roots(self):
        """List of root models in the document."""
    
    # Callback management
    def add_periodic_callback(self, callback, period_milliseconds):
        """
        Add a callback to be executed periodically.
        
        Parameters:
        - callback: Function to call periodically
        - period_milliseconds: Callback interval in milliseconds
        
        Returns:
        PeriodicCallback: Callback object for management
        """
    
    def add_timeout_callback(self, callback, timeout_milliseconds):
        """
        Add a callback to be executed once after a timeout.
        
        Parameters:
        - callback: Function to call after timeout
        - timeout_milliseconds: Timeout in milliseconds
        
        Returns:
        TimeoutCallback: Callback object for management
        """
    
    def add_next_tick_callback(self, callback):
        """Add a callback to be executed on the next tick."""
    
    def remove_periodic_callback(self, callback):
        """Remove a periodic callback."""
    
    def remove_timeout_callback(self, callback):
        """Remove a timeout callback."""
    
    def remove_next_tick_callback(self, callback):
        """Remove a next-tick callback."""
    
    # Property change callbacks
    def on_change(self, *callbacks):
        """Register callbacks for document changes."""
    
    # Session management  
    @property
    def session_context(self):
        """Current session context."""
    
    # Serialization
    def to_json_string(self, include_defaults=True):
        """Serialize document to JSON string."""
    
    def to_json(self, include_defaults=True):
        """Serialize document to JSON object."""
    
    # Title and metadata
    title: str             # Document title
    template: str          # HTML template name
    template_variables: Dict[str, Any]  # Template variables

class without_document_lock:
    """
    Context manager for operations outside document lock.
    
    Some operations (like network requests) should be performed
    outside the document lock to avoid blocking the server.
    """
    def __enter__(self): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

class ClientSession:
    """
    Client session for connecting to Bokeh server.
    
    Manages the connection between a client and a server application,
    handling authentication, document synchronization, and callbacks.
    """
    def __init__(self, session_id=None, websocket_url=None, io_loop=None, **kwargs):
        """
        Parameters:
        - session_id: Unique session identifier
        - websocket_url: WebSocket URL for server connection
        - io_loop: Tornado IOLoop instance
        """
    
    # Connection management
    def connect(self):
        """Connect to the server."""
    
    def close(self, why="closed"):
        """Close the connection."""
    
    @property  
    def connected(self):
        """Whether session is connected to server."""
    
    # Document access
    @property
    def document(self):
        """Document associated with this session."""
    
    # Callback management
    def on_change(self, *callbacks):
        """Register callbacks for session changes."""
    
    # Request handling
    def request_server_info(self):
        """Request server information."""
    
    def force_roundtrip(self):
        """Force a round-trip to synchronize with server."""

def pull_session(session_id=None, url=None, io_loop=None, **kwargs):
    """
    Pull session from server.
    
    Creates a ClientSession connected to an existing server session.
    
    Parameters:
    - session_id: Session ID to connect to (None for new session)
    - url: Server URL
    - io_loop: Tornado IOLoop instance
    
    Returns:
    ClientSession: Connected client session
    """

def push_session(document, session_id=None, url=None, io_loop=None, **kwargs):
    """
    Push session to server.
    
    Creates a new server session with the given document.
    
    Parameters:
    - document: Document to push to server
    - session_id: Session ID (None for auto-generated)
    - url: Server URL
    - io_loop: Tornado IOLoop instance
    
    Returns:
    ClientSession: Client session for the pushed document
    """

def show_session(session_id=None, server_url=None, session_url=None, 
                browser=None, new=None, **kwargs):
    """
    Display session in browser.
    
    Parameters:
    - session_id: Session ID to display
    - server_url: Base server URL
    - session_url: Full session URL
    - browser: Browser name/path
    - new: Browser window behavior ('tab', 'window', None)
    """

DEFAULT_SESSION_ID: str  # Default session identifier constant

class Server:
    """
    Bokeh server for hosting applications.
    
    The Server class provides the web server infrastructure for
    hosting Bokeh applications with WebSocket communication.
    """
    def __init__(self, applications, port=5006, address=None, **kwargs):
        """
        Parameters:
        - applications: Dict mapping URL paths to Application objects
        - port: Server port number
        - address: Server bind address
        """
    
    def start(self):
        """Start the server."""
    
    def stop(self):
        """Stop the server."""
    
    def run_until_shutdown(self):
        """Run server until shutdown signal."""
    
    @property
    def port(self):
        """Server port number."""
    
    @property
    def address(self):
        """Server bind address."""

# Server lifecycle callbacks
class SessionContext:
    """Context for server session lifecycle callbacks."""
    session_id: str
    server_context: 'ServerContext'
    destroyed: bool

class ServerContext:
    """Context for server application lifecycle callbacks."""
    application_context: 'ApplicationContext'
    sessions: Dict[str, SessionContext]

class PeriodicCallback:
    """Periodic callback that executes at regular intervals."""
    def __init__(self, callback, period, io_loop=None):
        """
        Parameters:
        - callback: Function to execute
        - period: Period in milliseconds
        - io_loop: Tornado IOLoop instance
        """
    
    def start(self):
        """Start the periodic callback."""
    
    def stop(self):
        """Stop the periodic callback."""
    
    @property
    def is_running(self):
        """Whether callback is currently running."""

class TimeoutCallback:
    """One-time callback that executes after a timeout."""
    def __init__(self, callback, timeout, io_loop=None):
        """
        Parameters:
        - callback: Function to execute
        - timeout: Timeout in milliseconds
        - io_loop: Tornado IOLoop instance
        """

class NextTickCallback:
    """Callback that executes on the next event loop tick."""
    def __init__(self, callback, io_loop=None):
        """
        Parameters:
        - callback: Function to execute
        - io_loop: Tornado IOLoop instance
        """

# save as: myapp.py
from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource, Button, Column
from bokeh.events import ButtonClick
import numpy as np

# Create data source
source = ColumnDataSource(data=dict(x=[1, 2, 3, 4], y=[1, 4, 2, 3]))

# Create plot
p = figure(width=400, height=400, title="Server Application")
line = p.line('x', 'y', source=source, line_width=3)

# Create button
button = Button(label="Update Data", button_type="success")

def update():
    """Update plot with random data."""
    n = np.random.randint(5, 10)
    new_data = dict(
        x=list(range(n)),
        y=np.random.random(n) * 10
    )
    source.data = new_data

# Handle button clicks
button.on_event(ButtonClick, lambda event: update())

# Create layout
layout = Column(button, p)

# Add to current document
curdoc().add_root(layout)
curdoc().title = "My Bokeh App"

# Run with: bokeh serve myapp.py

# Real-time data application
from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource
import numpy as np
from datetime import datetime
import random

# Create streaming data source
source = ColumnDataSource(data=dict(time=[], value=[]))

# Create plot
p = figure(width=600, height=400, title="Real-Time Data Stream",
           x_axis_type='datetime')
p.line('time', 'value', source=source, line_width=2)

def update_data():
    """Add new data point."""
    new_time = datetime.now()
    new_value = random.random() * 100
    
    # Stream new data (keep last 100 points)
    source.stream(dict(time=[new_time], value=[new_value]), rollover=100)

# Add periodic callback for updates every 500ms
curdoc().add_periodic_callback(update_data, 500)
curdoc().add_root(p)
curdoc().title = "Real-Time Dashboard"

from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource, Slider, Select, Column, Row
from bokeh.layouts import column, row
import numpy as np

# Data source
source = ColumnDataSource(data=dict(x=[], y=[]))

# Create plot
p = figure(width=500, height=400, title="Interactive Dashboard")
line = p.line('x', 'y', source=source, line_width=2)

# Widgets
freq_slider = Slider(start=0.1, end=5.0, value=1.0, step=0.1, 
                    title="Frequency")
func_select = Select(title="Function", value="sin", 
                    options=["sin", "cos", "tan"])
amplitude_slider = Slider(start=0.1, end=5.0, value=1.0, step=0.1,
                         title="Amplitude")

def update_plot():
    """Update plot based on widget values."""
    freq = freq_slider.value
    func_name = func_select.value
    amplitude = amplitude_slider.value
    
    x = np.linspace(0, 4*np.pi, 100)
    
    if func_name == "sin":
        y = amplitude * np.sin(freq * x)
    elif func_name == "cos":
        y = amplitude * np.cos(freq * x)
    else:  # tan
        y = amplitude * np.tan(freq * x)
        y = np.clip(y, -10, 10)  # Limit tan values
    
    source.data = dict(x=x, y=y)

# Widget callbacks
def on_change(attr, old, new):
    update_plot()

freq_slider.on_change('value', on_change)
func_select.on_change('value', on_change)
amplitude_slider.on_change('value', on_change)

# Initial plot
update_plot()

# Layout
controls = column(freq_slider, func_select, amplitude_slider)
layout = row(controls, p)

curdoc().add_root(layout)
curdoc().title = "Interactive Function Plotter"

from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource, Div, Column
import numpy as np
import uuid

# Generate unique session data
session_id = str(uuid.uuid4())[:8]
session_data = np.random.random(50) * 100

# Create data source with session-specific data
source = ColumnDataSource(data=dict(
    x=list(range(len(session_data))),
    y=session_data
))

# Create plot
p = figure(width=500, height=300, title=f"Session {session_id}")
p.line('x', 'y', source=source, line_width=2)
p.circle('x', 'y', source=source, size=8, alpha=0.7)

# Session info
info = Div(text=f"""
    <h3>Session Information</h3>
    <p><b>Session ID:</b> {session_id}</p>
    <p><b>Data Points:</b> {len(session_data)}</p>
    <p><b>Data Range:</b> {min(session_data):.2f} - {max(session_data):.2f}</p>
""")

layout = Column(info, p)
curdoc().add_root(layout)
curdoc().title = f"Multi-Session App - {session_id}"

# Optional: Add periodic callback to update data
def update_session_data():
    """Update session-specific data."""
    global session_data
    # Add some noise to existing data
    session_data = session_data + np.random.normal(0, 1, len(session_data))
    source.data = dict(x=list(range(len(session_data))), y=session_data)

curdoc().add_periodic_callback(update_session_data, 2000)  # Update every 2 seconds

# Client script to connect to server application
from bokeh.client import pull_session, push_session, show_session
from bokeh.plotting import figure
from bokeh.io import curdoc
import numpy as np

# Method 1: Pull existing session
session = pull_session(url="http://localhost:5006/myapp")
print(f"Connected to session: {session.session_id}")

# Access the document
doc = session.document
print(f"Document has {len(doc.roots)} root models")

# Method 2: Push new document to server
new_doc = curdoc()

# Create content for new document
p = figure(width=400, height=400)
x = np.linspace(0, 4*np.pi, 100)
p.line(x, np.sin(x))
new_doc.add_root(p)

# Push to server
push_session = push_session(new_doc, url="http://localhost:5006/")
print(f"Pushed document to session: {push_session.session_id}")

# Method 3: Show session in browser
show_session(session_id=push_session.session_id, 
            server_url="http://localhost:5006")

# application.py - Reusable application factory
from bokeh.application import Application
from bokeh.application.handlers import FunctionHandler
from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource, Slider, Column
import numpy as np

def create_app(config=None):
    """
    Factory function to create Bokeh application.
    
    Parameters:
    - config: Dictionary of configuration options
    """
    config = config or {}
    
    def modify_doc(doc):
        """Function to set up the document."""
        # Create plot based on config
        width = config.get('width', 500)
        height = config.get('height', 400)
        title = config.get('title', 'Default App')
        
        source = ColumnDataSource(data=dict(x=[], y=[]))
        
        p = figure(width=width, height=height, title=title)
        line = p.line('x', 'y', source=source, line_width=2)
        
        # Add interactivity
        slider = Slider(start=0.1, end=5.0, value=1.0, step=0.1,
                       title="Parameter")
        
        def update(attr, old, new):
            x = np.linspace(0, 4*np.pi, 100)
            y = np.sin(new * x)
            source.data = dict(x=x, y=y)
        
        slider.on_change('value', update)
        update(None, None, 1.0)  # Initial update
        
        layout = Column(slider, p)
        doc.add_root(layout)
        doc.title = title
    
    return Application(FunctionHandler(modify_doc))

# Usage:
# app1 = create_app({'title': 'Sin Wave', 'width': 600})
# app2 = create_app({'title': 'Different Config', 'height': 500})
# 
# # Run with multiple apps:
# # bokeh serve --show application.py:app1 application.py:app2

# Basic server commands

# Serve single application
bokeh serve myapp.py

# Serve on specific port
bokeh serve myapp.py --port 8080

# Serve multiple applications
bokeh serve app1.py app2.py --show

# Serve with auto-reload during development
bokeh serve myapp.py --dev

# Serve from directory
bokeh serve myapp/ --show

# Serve with custom URL prefix
bokeh serve myapp.py --prefix /custom/path

# Allow external connections
bokeh serve myapp.py --allow-websocket-origin=*

# Serve with authentication
bokeh serve myapp.py --oauth-provider=github