jupyter_server官方extension文档链接:https://jupyter-server.readthedocs.io/en/latest/developers/extensions.html

我们来做一个**真正“一用户一个连接”**的最小可运行示例，而且结构干净，不搞花活。

设计目标：

连接只存在于 Jupyter Server 进程
同一个用户的多个 notebook 共享
kernel 重启不影响连接
新 notebook 也能复用

思路是：

ServerApp 挂一个全局连接
提供一个 REST API
kernel 通过 HTTP 调用它

这是最稳的工程结构。

一、首先进行jupyterhub_config.py中的配置，指定jupyter_server_config.py的位置

c.Spawner.args = [
    "--config=/home/appuser/jupyterhub/config/jupyter_server_config.py"
]

下面是完整jupyter_server extension扩展包示例。

my_single_conn/
│
├── __init__.py
├── handlers.py
└── extension.py

二、extension.py

这个文件负责在 ServerApp 启动时注册连接和 handler。

# extension.py
"""
Jupyter Server Extension
实现：一个用户一个共享连接
连接生命周期绑定到 ServerApp
"""

from jupyter_server.utils import url_path_join
from .handlers import ConnHandler


class SharedConnection:
    """
    模拟一个共享连接对象
    你可以替换成 Spark / Socket / DB 等真实连接
    """

    def __init__(self):
        self.connected = False

    def connect(self):
        if not self.connected:
            print(">>> 建立真实连接")
            self.connected = True

    def status(self):
        return self.connected


def _jupyter_server_extension_points():
    """
    告诉 Jupyter 这是一个 server extension
    """
    return [{"module": "my_single_conn.extension"}]

import asyncio

async def monitor():
    while True:
        print("monitoring...")
        await asyncio.sleep(10)

# Jupyter Server 提供了一个简单的 API，用于从服务器扩展启动异步任务。这对于在事件循环运行后调用异步任务非常有用。

# 该函数应该被命名_start_jupyter_server_extension，并位于函数_load_jupyter_server_extension 旁边
async def _start_jupyter_server_extension(server_app):
    asyncio.create_task(monitor())


def _load_jupyter_server_extension(server_app):
    """
    在 Server 启动时执行
    """

    web_app = server_app.web_app
    host_pattern = ".*$"

    # 如果还没创建共享连接，则创建
    if "shared_conn" not in server_app.web_app.settings:
        server_app.log.info("初始化共享连接对象")
        server_app.web_app.settings["shared_conn"] = SharedConnection()

    # 注册 API handler
    base_url = web_app.settings["base_url"]
    route_pattern = url_path_join(base_url, "api", "shared-conn")

    web_app.add_handlers(host_pattern, [
        (route_pattern, ConnHandler),
    ])

    server_app.log.info("Shared connection extension loaded")

三、handlers.py

负责提供 API 接口。

# handlers.py
"""
处理前端或 kernel 的 HTTP 请求
"""

import json
from jupyter_server.base.handlers import APIHandler
from tornado import web


class ConnHandler(APIHandler):
    """
    GET /api/shared-conn
    返回连接状态
    POST /api/shared-conn
    确保连接建立
    """

    @web.authenticated
    def get(self):
        conn = self.application.settings["shared_conn"]
        self.finish(json.dumps({
            "connected": conn.status()
        }))

    @web.authenticated
    def post(self):
        conn = self.application.settings["shared_conn"]
        conn.connect()
        self.finish(json.dumps({
            "connected": conn.status()
        }))

四、__init__.py

from .extension import _jupyter_server_extension_points

五、启用 extension

在你的 jupyter_server_config.py 中加入：

import sys
sys.path.insert(0, "/home/appuser/liaiguo/jupyterhub/jupyter_extensions")

c.ServerApp.jpserver_extensions = {
    "my_single_conn": True
}

六、在jupyterlab的cell中调用接口

import requests
from jupyter_server.serverapp import list_running_servers

server = list(list_running_servers())

for s in server:
    try:
        r = requests.get(s["url"] + "api", timeout=0.2)
        if r.status_code == 200:
            server = s
            break
    except Exception:
        continue

base_url = server["url"].rstrip("/")
token = server["token"]

url = base_url + "/api/shared-conn"

headers = {
    "Authorization": f"token {token}"
}

# 建立连接
r = requests.post(url, headers=headers)
print("POST:", r.status_code, r.text)

# 查看状态
r = requests.get(url, headers=headers)
print("GET:", r.status_code, r.text)