tessl/pypi-jupyter-server

The backend infrastructure for Jupyter web applications providing core services, APIs, and REST endpoints.

Overview

Eval results

Files

Services

Name: tessl/pypi-jupyter-server
Author: tessl

The Jupyter Server provides a comprehensive set of service modules that handle core functionality like file management, kernel orchestration, session tracking, and API endpoints.

Contents Service

Manages file and notebook operations with support for different storage backends.

from jupyter_server.services.contents.manager import ContentsManager, AsyncContentsManager
from jupyter_server.services.contents.filemanager import FileContentsManager, AsyncFileContentsManager
from jupyter_server.services.contents.largefilemanager import LargeFileManager, AsyncLargeFileManager
from jupyter_server.services.contents.checkpoints import Checkpoints, FileCheckpoints, AsyncCheckpoints, AsyncFileCheckpoints

ContentsManager

Abstract base class for content management systems.

from jupyter_server.services.contents.manager import ContentsManager

class MyContentsManager(ContentsManager):
    """Custom contents manager implementation."""

    def get(self, path: str, content: bool = True, type: str = None, format: str = None):
        """Get a file or directory model."""
        # Return file/directory model dictionary
        return {
            "name": "example.ipynb",
            "path": path,
            "type": "notebook",
            "format": "json",
            "content": notebook_content if content else None,
            "size": 1024,
            "last_modified": datetime.now().isoformat(),
            "created": datetime.now().isoformat(),
            "mimetype": "application/x-ipynb+json",
        }

    def save(self, model: dict, path: str):
        """Save a file or directory."""
        # Save the model and return updated model
        return self.get(path)

    def delete_file(self, path: str):
        """Delete a file."""
        # Implement file deletion
        pass

    def rename_file(self, old_path: str, new_path: str):
        """Rename a file."""
        # Implement file renaming
        pass

    def file_exists(self, path: str) -> bool:
        """Check if file exists."""
        return True  # Implement existence check

    def dir_exists(self, path: str) -> bool:
        """Check if directory exists."""
        return True  # Implement directory check

    def is_hidden(self, path: str) -> bool:
        """Check if path is hidden."""
        return path.startswith('.')

FileContentsManager

File system-based implementation.

from jupyter_server.services.contents.filemanager import FileContentsManager

# Configure file contents manager
contents_manager = FileContentsManager()
contents_manager.root_dir = "/path/to/notebooks"
contents_manager.hide_globs = ["__pycache__", "*.pyc", ".git"]
contents_manager.untitled_notebook = "Untitled.ipynb"
contents_manager.untitled_file = "untitled.txt"
contents_manager.untitled_directory = "Untitled Folder"

# Get file/directory listing
model = contents_manager.get("")  # Root directory
print(f"Found {len(model['content'])} items")

# Save notebook
notebook_model = {
    "type": "notebook",
    "format": "json",
    "content": {
        "cells": [],
        "metadata": {},
        "nbformat": 4,
        "nbformat_minor": 5,
    }
}
saved = contents_manager.save(notebook_model, "new_notebook.ipynb")

AsyncContentsManager

Asynchronous version of ContentsManager for better performance in async contexts.

from jupyter_server.services.contents.manager import AsyncContentsManager
import asyncio

class MyAsyncContentsManager(AsyncContentsManager):
    """Custom async contents manager implementation."""

    async def get(self, path: str, content: bool = True, type: str = None, format: str = None):
        """Get a file or directory model asynchronously."""
        # Async file operations here
        await asyncio.sleep(0)  # Simulate async operation
        return {
            "name": "example.ipynb",
            "path": path,
            "type": "notebook",
            "format": "json",
            "content": notebook_content if content else None,
        }

    async def save(self, model: dict, path: str):
        """Save a file or directory asynchronously."""
        # Async save operations here
        await asyncio.sleep(0)  # Simulate async operation
        return await self.get(path)

# Usage
async def example_usage():
    contents_manager = MyAsyncContentsManager()
    model = await contents_manager.get("example.ipynb")
    print(f"Got notebook: {model['name']}")

LargeFileManager

Optimized contents manager for handling large files efficiently.

from jupyter_server.services.contents.largefilemanager import LargeFileManager

# Configure for large file handling
large_file_manager = LargeFileManager()
large_file_manager.root_dir = "/path/to/large/files"
large_file_manager.max_file_size_to_upload = 10 * 1024 * 1024 * 1024  # 10GB

# Use for large file operations
large_file = large_file_manager.get("huge_dataset.csv", content=False)
print(f"Large file size: {large_file['size']} bytes")

Checkpoints

Version control and backup functionality for files.

from jupyter_server.services.contents.checkpoints import (
    Checkpoints,
    FileCheckpoints,
    GenericCheckpointsMixin,
)

class MyCheckpoints(Checkpoints):
    """Custom checkpoint implementation."""

    def create_checkpoint(self, contents_mgr, path: str) -> dict:
        """Create a checkpoint."""
        checkpoint_id = f"checkpoint-{int(time.time())}"
        return {
            "id": checkpoint_id,
            "last_modified": datetime.now().isoformat(),
        }

    def restore_checkpoint(self, contents_mgr, checkpoint_id: str, path: str):
        """Restore from checkpoint."""
        # Implement checkpoint restoration
        pass

    def delete_checkpoint(self, checkpoint_id: str, path: str):
        """Delete a checkpoint."""
        # Implement checkpoint deletion
        pass

    def list_checkpoints(self, path: str) -> list:
        """List available checkpoints."""
        return [
            {
                "id": "checkpoint-123456789",
                "last_modified": datetime.now().isoformat(),
            }
        ]

# Use with file contents manager
contents_manager = FileContentsManager()
contents_manager.checkpoints = MyCheckpoints()

# Create checkpoint
checkpoint = contents_manager.create_checkpoint("example.ipynb")
print(f"Created checkpoint: {checkpoint['id']}")

# List checkpoints
checkpoints = contents_manager.list_checkpoints("example.ipynb")
print(f"Available checkpoints: {len(checkpoints)}")

Contents Utilities

from jupyter_server.services.contents.fileio import atomic_writing, copy2_safe

# Safe atomic file writing
with atomic_writing("/path/to/file.txt") as f:
    f.write("Content that will be written atomically")

# Safe file copying
copy2_safe("/source/file.txt", "/dest/file.txt")

Kernels Service

Manages Jupyter kernel lifecycles and communication.

from jupyter_server.services.kernels.kernelmanager import MappingKernelManager, AsyncMappingKernelManager

MappingKernelManager

Manages multiple kernel instances.

from jupyter_server.services.kernels.kernelmanager import MappingKernelManager
from jupyter_client.kernelspec import KernelSpecManager

# Create kernel manager
kernel_spec_manager = KernelSpecManager()
kernel_manager = MappingKernelManager(
    kernel_spec_manager=kernel_spec_manager
)

# Start a kernel
kernel_id = await kernel_manager.start_kernel(kernel_name="python3")
print(f"Started kernel: {kernel_id}")

# Get kernel info
kernel = kernel_manager.get_kernel(kernel_id)
print(f"Kernel status: {kernel.execution_state}")

# List running kernels
kernels = kernel_manager.list_kernels()
for k in kernels:
    print(f"Kernel {k['id']}: {k['name']} ({k['execution_state']})")

# Interrupt kernel
await kernel_manager.interrupt_kernel(kernel_id)

# Restart kernel
await kernel_manager.restart_kernel(kernel_id)

# Shutdown kernel
await kernel_manager.shutdown_kernel(kernel_id)

AsyncMappingKernelManager

Asynchronous version for better performance.

from jupyter_server.services.kernels.kernelmanager import AsyncMappingKernelManager
import asyncio

async def manage_kernels():
    kernel_manager = AsyncMappingKernelManager()

    # Start multiple kernels
    kernel_ids = []
    for i in range(3):
        kernel_id = await kernel_manager.start_kernel(kernel_name="python3")
        kernel_ids.append(kernel_id)
        print(f"Started kernel {i+1}: {kernel_id}")

    # Execute code in kernels
    for kernel_id in kernel_ids:
        kernel = kernel_manager.get_kernel(kernel_id)
        msg_id = kernel.execute("print('Hello from kernel!')")
        print(f"Executed in kernel {kernel_id}: {msg_id}")

    # Shutdown all kernels
    for kernel_id in kernel_ids:
        await kernel_manager.shutdown_kernel(kernel_id)

# Run async function
asyncio.run(manage_kernels())

Sessions Service

Manages the relationship between kernels and frontend clients.

from jupyter_server.services.sessions.sessionmanager import SessionManager

SessionManager

Tracks kernel-frontend session relationships.

from jupyter_server.services.sessions.sessionmanager import SessionManager
from jupyter_server.services.kernels.kernelmanager import MappingKernelManager

# Create session manager
kernel_manager = MappingKernelManager()
session_manager = SessionManager(kernel_manager=kernel_manager)

# Create a session
session = await session_manager.create_session(
    path="notebook.ipynb",
    type="notebook",
    name="Python 3",
    kernel_name="python3",
)

print(f"Created session: {session['id']}")
print(f"Session path: {session['path']}")
print(f"Kernel ID: {session['kernel']['id']}")

# List all sessions
sessions = session_manager.list_sessions()
for s in sessions:
    print(f"Session {s['id']}: {s['path']} -> {s['kernel']['id']}")

# Get session by ID
session = session_manager.get_session(session_id=session['id'])

# Update session (change path/name)
updated_session = await session_manager.update_session(
    session_id=session['id'],
    path="renamed_notebook.ipynb",
    type="notebook",
    name="Python 3 (Updated)",
)

# Delete session (also shuts down kernel)
await session_manager.delete_session(session_id=session['id'])

Session Records

from jupyter_server.services.sessions.sessionmanager import KernelSessionRecord

# Session record tracking
record = KernelSessionRecord(
    kernel_id="kernel-123",
    session_id="session-456",
    kernel_name="python3",
)

print(f"Kernel: {record.kernel_id}")
print(f"Session: {record.session_id}")
print(f"Name: {record.kernel_name}")

Kernel Specs Service

Manages available kernel specifications.

from jupyter_server.services.kernelspecs.handlers import kernelspec_model

Kernel Spec Operations

from jupyter_server.services.kernelspecs.handlers import (
    kernelspec_model,
    is_kernelspec_model,
)
from jupyter_client.kernelspec import KernelSpecManager

# Create kernel spec manager
spec_manager = KernelSpecManager()

# Get all kernel specs
specs = spec_manager.get_all_specs()
for name, spec in specs.items():
    print(f"Kernel: {name}")
    print(f"  Display name: {spec['spec']['display_name']}")
    print(f"  Language: {spec['spec']['language']}")

# Get specific kernel spec
python_spec = spec_manager.get_kernel_spec("python3")
print(f"Python kernel: {python_spec.display_name}")

# Create kernel spec model for API
model = kernelspec_model(spec_manager, "python3")
print(f"Model: {model}")

# Validate kernel spec model
is_valid = is_kernelspec_model(model)
print(f"Valid model: {is_valid}")

Configuration Service

Provides runtime configuration management.

from jupyter_server.services.config.manager import ConfigManager

ConfigManager

Manages JSON configuration files and runtime settings.

from jupyter_server.services.config.manager import ConfigManager
import json

# Create config manager
config_manager = ConfigManager()

# Get configuration section
config = config_manager.get("my_extension")
print(f"Extension config: {config}")

# Set configuration value
config_manager.set("my_extension", "setting", "value")

# Update multiple settings
config_manager.update("my_extension", {
    "setting1": "value1",
    "setting2": 42,
    "setting3": True,
})

# Get updated config
updated_config = config_manager.get("my_extension")
print(f"Updated config: {json.dumps(updated_config, indent=2)}")

API Service

Provides OpenAPI specifications and server status endpoints.

from jupyter_server.services.api.handlers import APISpecHandler, APIStatusHandler

API Information

# Get API specification (handled by APISpecHandler)
# GET /api/spec.yaml returns OpenAPI spec

# Get server status (handled by APIStatusHandler)
# GET /api/status returns:
{
    "started": "2023-01-01T00:00:00.000000Z",
    "last_activity": "2023-01-01T01:00:00.000000Z",
    "kernels": 2,
    "connections": 5,
}

# Get user identity (handled by IdentityHandler)
# GET /api/me returns:
{
    "username": "user",
    "name": "User Name",
    "display_name": "User Name",
    "initials": "UN",
    "color": "#orange",
}

Events Service

Handles server-side events and WebSocket subscriptions.

from jupyter_server.services.events.handlers import EventHandler, SubscribeWebsocket

Event Handling

# Events are handled through WebSocket connections
# GET /api/events/subscribe opens WebSocket

# Event message format:
{
    "schema_id": "https://events.jupyter.org/jupyter_server/session/v1",
    "version": 1,
    "timestamp": "2023-01-01T00:00:00.000Z",
    "data": {
        "session_id": "session-123",
        "action": "started",
    }
}

NBConvert Service

Provides notebook conversion capabilities.

from jupyter_server.services.nbconvert.handlers import NbconvertRootHandler

Notebook Conversion

# Convert notebook via API
# POST /api/nbconvert/format/path/to/notebook.ipynb

# Request body:
{
    "output_format": "html",  # html, pdf, latex, etc.
    "options": {
        "template": "classic",
        "exclude_input": False,
    }
}

# Response: converted content or download link

Security Service

Content Security Policy and security reporting.

from jupyter_server.services.security.handlers import CSPReportHandler

CSP Reporting

# CSP violation reports are sent to:
# POST /api/security/csp-report

# Report format:
{
    "csp-report": {
        "blocked-uri": "eval",
        "document-uri": "http://localhost:8888/",
        "original-policy": "default-src 'self'",
        "violated-directive": "script-src",
    }
}

Custom Service Implementation

Creating a Custom Service

from jupyter_server.base.handlers import APIHandler
from jupyter_server.auth.decorator import authorized
from tornado import web
import json

class MyCustomService:
    """Custom service implementation."""

    def __init__(self):
        self.data = {}

    async def store_data(self, key: str, value):
        """Store data in service."""
        self.data[key] = value
        return {"success": True, "key": key}

    async def get_data(self, key: str = None):
        """Retrieve data from service."""
        if key:
            return self.data.get(key)
        return self.data

    async def delete_data(self, key: str):
        """Delete data from service."""
        if key in self.data:
            del self.data[key]
            return {"success": True, "key": key}
        raise web.HTTPError(404, f"Key '{key}' not found")

class MyServiceHandler(APIHandler):
    """API handler for custom service."""

    def initialize(self, service):
        self.service = service

    @authorized
    async def get(self, key=None):
        """Get data from service."""
        try:
            data = await self.service.get_data(key)
            self.finish({"data": data})
        except Exception as e:
            raise web.HTTPError(500, str(e))

    @authorized
    async def post(self):
        """Store data in service."""
        try:
            body = json.loads(self.request.body)
            key = body.get("key")
            value = body.get("value")

            if not key:
                raise web.HTTPError(400, "Missing 'key' parameter")

            result = await self.service.store_data(key, value)
            self.finish(result)
        except json.JSONDecodeError:
            raise web.HTTPError(400, "Invalid JSON")
        except Exception as e:
            raise web.HTTPError(500, str(e))

    @authorized
    async def delete(self, key):
        """Delete data from service."""
        try:
            result = await self.service.delete_data(key)
            self.finish(result)
        except Exception as e:
            if isinstance(e, web.HTTPError):
                raise
            raise web.HTTPError(500, str(e))

# Register service handlers
def register_custom_service(web_app):
    """Register custom service with server."""
    service = MyCustomService()

    handlers = [
        (r"/api/custom", MyServiceHandler, {"service": service}),
        (r"/api/custom/(.+)", MyServiceHandler, {"service": service}),
    ]

    web_app.add_handlers(".*$", handlers)

Service Integration

Integrating Services with Extensions

from jupyter_server.extension.application import ExtensionApp

## Events Service

Handles event logging and tracking across the Jupyter server.

```python{ .api }
from jupyter_server.services.events import EventLogger
from jupyter_events import event_logger

# Create event logger
event_logger = EventLogger()

# Log custom events
event_logger.emit(
    schema_id="https://events.jupyter.org/jupyter_server",
    version=1,
    event={
        "action": "server_start",
        "timestamp": datetime.now().isoformat(),
        "details": {"port": 8888}
    }
)

# Register event schema
event_logger.register_event_schema({
    "schema_id": "https://my-app.com/events",
    "version": 1,
    "properties": {
        "action": {"type": "string"},
        "user_id": {"type": "string"}
    }
})

Kernelspecs Service

Manages available kernel specifications and types.

from jupyter_server.services.kernelspecs import KernelSpecManager
from jupyter_client.kernelspec import KernelSpecManager as ClientKernelSpecManager

# Get kernel specs manager
kernelspec_manager = KernelSpecManager()

# List available kernel specs
specs = kernelspec_manager.get_all_specs()
for name, spec in specs.items():
    print(f"Kernel: {name}")
    print(f"  Display name: {spec['spec']['display_name']}")
    print(f"  Language: {spec['spec']['language']}")

# Get specific kernel spec
python_spec = kernelspec_manager.get_kernel_spec("python3")
print(f"Python kernel command: {python_spec.argv}")

Extension Services Integration

Custom services can be integrated through extensions.

from jupyter_server.extension.application import ExtensionApp

class MyExtensionWithServices(ExtensionApp):
    name = "my_service_extension"

    def initialize_services(self):
        """Initialize custom services."""
        # Create custom service
        self.custom_service = MyCustomService()

        # Configure existing services
        self.contents_manager.hide_globs.append("*.tmp")

    def initialize_handlers(self):
        """Register service handlers."""
        handlers = [
            (
                r"/my_extension/api/data",
                MyServiceHandler,
                {"service": self.custom_service}
            ),
        ]
        self.handlers.extend(handlers)

    def initialize_settings(self):
        """Update server settings."""
        self.settings.update({
            "my_custom_service": self.custom_service,
        })

Service Configuration

# In jupyter_server_config.py

c = get_config()

# Configure contents manager
c.ServerApp.contents_manager_class = 'my_package.MyContentsManager'
c.MyContentsManager.root_dir = '/custom/notebook/path'
c.MyContentsManager.hide_globs = ['*.tmp', '__pycache__']

# Configure kernel manager
c.ServerApp.kernel_manager_class = 'my_package.MyKernelManager'
c.MyKernelManager.default_kernel_name = 'python3'

# Configure session manager
c.ServerApp.session_manager_class = 'my_package.MySessionManager'
c.MySessionManager.session_cleanup_interval = 300  # 5 minutes

Install with Tessl CLI