Configuration Management

from jupyter_server.config_manager import BaseJSONConfigManager

from jupyter_server.config_manager import BaseJSONConfigManager
import os

class MyConfigManager(BaseJSONConfigManager):
    """Custom configuration manager."""

    def __init__(self, config_dir=None):
        """Initialize with configuration directory."""
        if config_dir is None:
            config_dir = os.path.expanduser("~/.jupyter")

        super().__init__(config_dir=config_dir)

# Create config manager
config_manager = MyConfigManager()

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

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

# Update multiple settings
config_manager.update("my_extension", {
    "host": "localhost",
    "port": 8888,
    "debug": True,
    "allowed_users": ["alice", "bob"],
})

# Get updated configuration
updated_config = config_manager.get("my_extension")
print(f"Updated config: {updated_config}")

from jupyter_server.services.config.manager import ConfigManager

# Service-level configuration manager
service_config = ConfigManager()

# Get service configuration
jupyter_config = service_config.get("jupyter")
notebook_config = service_config.get("notebook")

# Update service configuration
service_config.update("jupyter", {
    "kernel_timeout": 60,
    "shutdown_timeout": 5,
})

# Atomic configuration updates
with service_config.config_transaction("my_service") as config:
    config["setting1"] = "value1"
    config["setting2"] = "value2"
    # Changes committed when context exits

from jupyter_server.config_manager import recursive_update, remove_defaults

from jupyter_server.config_manager import recursive_update, remove_defaults

# Deep dictionary merging
base_config = {
    "server": {
        "host": "localhost",
        "port": 8888,
        "ssl": {"enabled": False}
    },
    "logging": {"level": "INFO"}
}

update_config = {
    "server": {
        "port": 9999,  # Override port
        "ssl": {"enabled": True, "cert_file": "/path/cert.pem"}  # Merge SSL config
    },
    "extensions": {"my_ext": True}  # Add new section
}

merged = recursive_update(base_config, update_config)
print(merged)
# {
#     "server": {
#         "host": "localhost",
#         "port": 9999,
#         "ssl": {"enabled": True, "cert_file": "/path/cert.pem"}
#     },
#     "logging": {"level": "INFO"},
#     "extensions": {"my_ext": True}
# }

# Remove default values
defaults = {
    "server": {"host": "localhost", "port": 8888},
    "logging": {"level": "INFO"}
}

config = {
    "server": {"host": "localhost", "port": 9999},
    "logging": {"level": "INFO"}
}

cleaned = remove_defaults(config, defaults)
print(cleaned)
# {"server": {"port": 9999}}  # Only non-default values remain

from jupyter_server.traittypes import TypeFromClasses, InstanceFromClasses

from jupyter_server.traittypes import TypeFromClasses, InstanceFromClasses
from traitlets import HasTraits, Unicode, Int, Bool, List
from jupyter_server.auth.identity import IdentityProvider
from jupyter_server.auth.authorizer import Authorizer

class MyServerConfig(HasTraits):
    """Custom server configuration with validation."""

    # Basic traits
    host = Unicode("localhost", help="Server host address").tag(config=True)
    port = Int(8888, help="Server port").tag(config=True)
    debug = Bool(False, help="Enable debug mode").tag(config=True)

    # List of allowed origins
    allowed_origins = List(
        trait=Unicode(),
        default_value=["localhost"],
        help="List of allowed origins for CORS"
    ).tag(config=True)

    # Type validation - must be subclass of IdentityProvider
    identity_provider_class = TypeFromClasses(
        default_value=IdentityProvider,
        klasses=[IdentityProvider],
        help="Identity provider class"
    ).tag(config=True)

    # Instance validation - must be instance of Authorizer
    authorizer = InstanceFromClasses(
        klasses=[Authorizer],
        allow_none=True,
        help="Authorizer instance"
    ).tag(config=True)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Validate configuration after initialization
        self._validate_config()

    def _validate_config(self):
        """Custom configuration validation."""
        if self.port < 1 or self.port > 65535:
            raise ValueError(f"Invalid port: {self.port}")

        if self.debug and "0.0.0.0" in self.host:
            self.log.warning("Debug mode enabled with public host - security risk!")

# Usage
config = MyServerConfig()
config.host = "127.0.0.1"
config.port = 9999
config.debug = True
config.allowed_origins = ["localhost", "127.0.0.1"]

print(f"Server will run on {config.host}:{config.port}")

from jupyter_server.serverapp import ServerApp
from traitlets.config.loader import PyFileConfigLoader

# Load configuration from Python file
def load_config_file(config_path):
    """Load configuration from Python file."""
    loader = PyFileConfigLoader("jupyter_server_config.py", config_path)
    config = loader.load_config()
    return config

# Apply configuration to server
app = ServerApp()

# Load from file
config = load_config_file("/path/to/config")
app.update_config(config)

# Or load directly
app.config_file = "jupyter_server_config.py"
app.config_dir = "/path/to/config/dir"
app.load_config_file()

# Command line override
app.parse_command_line([
    "--ServerApp.port=9999",
    "--ServerApp.host=0.0.0.0",
    "--ServerApp.debug=True"
])

from jupyter_server.extension.config import ExtensionConfigManager
from jupyter_server.extension.application import ExtensionApp
from traitlets import Unicode, Int, Bool, List

class MyExtensionConfig(ExtensionApp):
    """Extension with custom configuration."""

    name = "my_extension"

    # Extension-specific configuration
    api_endpoint = Unicode(
        "/my_extension/api",
        help="API endpoint for extension"
    ).tag(config=True)

    max_connections = Int(
        100,
        help="Maximum number of connections"
    ).tag(config=True)

    enable_caching = Bool(
        True,
        help="Enable response caching"
    ).tag(config=True)

    allowed_methods = List(
        trait=Unicode(),
        default_value=["GET", "POST"],
        help="Allowed HTTP methods"
    ).tag(config=True)

    def initialize_settings(self):
        """Initialize extension settings."""
        self.settings.update({
            "my_extension_config": {
                "api_endpoint": self.api_endpoint,
                "max_connections": self.max_connections,
                "enable_caching": self.enable_caching,
                "allowed_methods": self.allowed_methods,
            }
        })

# Configuration in jupyter_server_config.py
c = get_config()
c.MyExtensionConfig.api_endpoint = "/custom/api"
c.MyExtensionConfig.max_connections = 200
c.MyExtensionConfig.enable_caching = False

# Or using extension config manager
ext_config = ExtensionConfigManager(extension_name="my_extension")
ext_config.update("MyExtensionConfig", {
    "api_endpoint": "/custom/api",
    "max_connections": 200,
    "enable_caching": False,
})

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

class DynamicConfigHandler(APIHandler):
    """Handler for runtime configuration updates."""

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

    @authorized
    async def get(self, section=None):
        """Get configuration."""
        if section:
            config = self.config_manager.get(section)
        else:
            # Get all configuration sections
            config = {
                section: self.config_manager.get(section)
                for section in self.config_manager.list_sections()
            }

        self.finish({"config": config})

    @authorized
    async def put(self, section):
        """Update configuration."""
        try:
            data = self.get_json_body()
            self.config_manager.update(section, data)

            # Get updated config
            updated_config = self.config_manager.get(section)
            self.finish({"config": updated_config})

        except Exception as e:
            raise web.HTTPError(400, f"Configuration update failed: {e}")

    @authorized
    async def delete(self, section):
        """Delete configuration section."""
        try:
            self.config_manager.delete_section(section)
            self.set_status(204)
            self.finish()

        except Exception as e:
            raise web.HTTPError(400, f"Configuration deletion failed: {e}")

# Register configuration API
handlers = [
    (r"/api/config", DynamicConfigHandler, {"config_manager": config_manager}),
    (r"/api/config/(.+)", DynamicConfigHandler, {"config_manager": config_manager}),
]

from traitlets import HasTraits, TraitError, validate
import jsonschema

class ValidatedConfig(HasTraits):
    """Configuration with JSON schema validation."""

    def __init__(self, **kwargs):
        # Define JSON schema for validation
        self.schema = {
            "type": "object",
            "properties": {
                "host": {"type": "string", "format": "hostname"},
                "port": {"type": "integer", "minimum": 1, "maximum": 65535},
                "ssl_options": {
                    "type": "object",
                    "properties": {
                        "certfile": {"type": "string"},
                        "keyfile": {"type": "string"},
                    },
                    "required": ["certfile", "keyfile"]
                }
            },
            "required": ["host", "port"]
        }

        super().__init__(**kwargs)

    @validate('config_data')
    def _validate_config_data(self, proposal):
        """Validate configuration against JSON schema."""
        try:
            jsonschema.validate(proposal['value'], self.schema)
            return proposal['value']
        except jsonschema.ValidationError as e:
            raise TraitError(f"Configuration validation failed: {e.message}")

    def validate_and_load(self, config_dict):
        """Validate and load configuration."""
        # Validate against schema
        jsonschema.validate(config_dict, self.schema)

        # Apply configuration
        for key, value in config_dict.items():
            if hasattr(self, key):
                setattr(self, key, value)

# Usage
config = ValidatedConfig()
try:
    config.validate_and_load({
        "host": "localhost",
        "port": 8888,
        "ssl_options": {
            "certfile": "/path/to/cert.pem",
            "keyfile": "/path/to/key.pem"
        }
    })
except jsonschema.ValidationError as e:
    print(f"Invalid configuration: {e.message}")

from traitlets import HasTraits, Unicode, Int, Bool
from traitlets.config.configurable import Configurable
import os

class EnvironmentConfig(Configurable):
    """Configuration with environment variable support."""

    host = Unicode(
        help="Server host (env: JUPYTER_HOST)"
    ).tag(config=True, env="JUPYTER_HOST")

    port = Int(
        8888,
        help="Server port (env: JUPYTER_PORT)"
    ).tag(config=True, env="JUPYTER_PORT")

    debug = Bool(
        False,
        help="Debug mode (env: JUPYTER_DEBUG)"
    ).tag(config=True, env="JUPYTER_DEBUG")

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._load_from_env()

    def _load_from_env(self):
        """Load configuration from environment variables."""
        # Get traits with environment variable tags
        for trait_name, trait in self.class_traits().items():
            env_var = trait.metadata.get('env')
            if env_var and env_var in os.environ:
                env_value = os.environ[env_var]

                # Convert environment string to appropriate type
                if isinstance(trait, Int):
                    setattr(self, trait_name, int(env_value))
                elif isinstance(trait, Bool):
                    setattr(self, trait_name, env_value.lower() in ('true', '1', 'yes', 'on'))
                else:
                    setattr(self, trait_name, env_value)

# Usage with environment variables
# JUPYTER_HOST=0.0.0.0 JUPYTER_PORT=9999 JUPYTER_DEBUG=true python server.py

config = EnvironmentConfig()
print(f"Host: {config.host}")     # From JUPYTER_HOST
print(f"Port: {config.port}")     # From JUPYTER_PORT
print(f"Debug: {config.debug}")   # From JUPYTER_DEBUG

# jupyter_server_config.py

c = get_config()

# Basic server settings
c.ServerApp.ip = '0.0.0.0'
c.ServerApp.port = 8888
c.ServerApp.open_browser = False
c.ServerApp.allow_remote_access = True

# Authentication
c.ServerApp.token = 'your-secret-token'
c.ServerApp.password = 'sha1:...'
c.ServerApp.allow_password_change = True

# Directories
c.ServerApp.root_dir = '/data/notebooks'
c.ServerApp.preferred_dir = '/data/notebooks/work'

# Security
c.ServerApp.allow_origin = '*'
c.ServerApp.allow_credentials = True
c.ServerApp.disable_check_xsrf = False

# SSL/TLS
c.ServerApp.certfile = '/path/to/cert.pem'
c.ServerApp.keyfile = '/path/to/key.pem'

# Logging
c.ServerApp.log_level = 'INFO'

# Custom managers
c.ServerApp.contents_manager_class = 'my_package.MyContentsManager'
c.ServerApp.kernel_manager_class = 'my_package.MyKernelManager'
c.ServerApp.session_manager_class = 'my_package.MySessionManager'

# Identity and authorization
c.ServerApp.identity_provider_class = 'my_package.MyIdentityProvider'
c.ServerApp.authorizer_class = 'my_package.MyAuthorizer'

# Extensions
c.ServerApp.jpserver_extensions = {
    'my_extension': True,
    'another_extension': True,
}

# Extension-specific configuration
c.MyExtension.setting = 'value'
c.MyExtension.enabled_features = ['feature1', 'feature2']

# Contents manager configuration
c.MyContentsManager.hide_globs = ['__pycache__', '*.pyc', '.git']
c.MyContentsManager.untitled_notebook = 'Untitled.ipynb'
c.MyContentsManager.untitled_file = 'untitled.txt'

# Kernel manager configuration
c.MyKernelManager.default_kernel_name = 'python3'
c.MyKernelManager.kernel_info_timeout = 60
c.MyKernelManager.shutdown_wait_time = 5.0

# Custom authorizer configuration
c.MyAuthorizer.admin_users = ['admin', 'superuser']
c.MyAuthorizer.read_only_users = ['guest', 'viewer']
c.MyAuthorizer.allowed_paths = ['/public', '/shared']

from jupyter_server.serverapp import ServerApp
from traitlets.config.loader import PyFileConfigLoader
import os

class MultiEnvConfig:
    """Configuration manager for multiple environments."""

    def __init__(self, config_dir=None):
        self.config_dir = config_dir or os.path.expanduser("~/.jupyter")
        self.environment = os.environ.get("JUPYTER_ENV", "development")

    def load_config(self):
        """Load environment-specific configuration."""
        # Base configuration
        base_config = self._load_config_file("jupyter_server_config.py")

        # Environment-specific configuration
        env_config_file = f"jupyter_server_config_{self.environment}.py"
        env_config = self._load_config_file(env_config_file)

        # Merge configurations
        if env_config:
            base_config.merge(env_config)

        return base_config

    def _load_config_file(self, filename):
        """Load configuration from file."""
        config_path = os.path.join(self.config_dir, filename)
        if os.path.exists(config_path):
            loader = PyFileConfigLoader(filename, self.config_dir)
            return loader.load_config()
        return None

    def create_server_app(self):
        """Create server app with environment configuration."""
        app = ServerApp()
        config = self.load_config()
        if config:
            app.update_config(config)
        return app

# Usage
# JUPYTER_ENV=production python server.py
config_manager = MultiEnvConfig()
app = config_manager.create_server_app()
app.initialize()
app.start()

import json
import time
from jupyter_server.services.config.manager import ConfigManager
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ConfigFileHandler(FileSystemEventHandler):
    """Monitor configuration file changes."""

    def __init__(self, config_manager):
        self.config_manager = config_manager
        self.last_reload = time.time()

    def on_modified(self, event):
        """Handle configuration file modification."""
        if event.is_directory:
            return

        if event.src_path.endswith('.json'):
            # Debounce rapid changes
            if time.time() - self.last_reload < 1.0:
                return

            self.reload_config(event.src_path)
            self.last_reload = time.time()

    def reload_config(self, config_path):
        """Reload configuration from file."""
        try:
            with open(config_path, 'r') as f:
                new_config = json.load(f)

            # Extract section name from filename
            section = os.path.basename(config_path).replace('.json', '')

            # Update configuration
            self.config_manager.update(section, new_config)
            print(f"Reloaded configuration for {section}")

        except Exception as e:
            print(f"Failed to reload config from {config_path}: {e}")

# Setup configuration monitoring
config_manager = ConfigManager()
event_handler = ConfigFileHandler(config_manager)
observer = Observer()
observer.schedule(event_handler, path="/path/to/config", recursive=False)
observer.start()

# Configuration will automatically reload when files change