如何将Bokeh Server集成到Django 实现鉴权、数据存储与权限管控

以下实现完全覆盖所列需求，基于server_document()完成集成，无需修改Bokeh核心源码。

核心架构逻辑 #

所有认证、权限、数据层逻辑统一放在Django侧实现，Bokeh Server仅负责交互计算与图表渲染，不直接对公网暴露端口，所有Bokeh请求经Django反向代理转发，从架构上避免权限绕过。

分步实现 #

• 基础环境准备
  安装依赖：
  pip install django bokeh sqlalchemy pandas
  数据库直接复用Django的业务库，提前建好核心表（可通过Django ORM创建，也可直接写SQL建表）：

  • Django自带用户认证表（直接复用django.contrib.auth模块，无需自行开发）
  • 用户应用权限表：存储用户ID、可访问的Bokeh应用名
  • 公共数据表：存储成本类等全用户通用的数据
  • 用户运算结果表：存储用户ID、计算结果、创建时间
  • 用户会话状态表：存储用户ID、会话状态JSON、更新时间
    Bokeh启动时添加参数--allow-websocket-origin=你的Django服务域名 --address=127.0.0.1，确保仅本地Django服务能访问Bokeh端口，公网无法直接连接。

• Django侧嵌入逻辑（对应server_document()集成）
  编写专用路由视图处理Bokeh页面嵌入，依次完成登录校验、权限校验、生成认证令牌、返回嵌入脚本，最小代码示例：

  from django.shortcuts import render, HttpResponseForbidden, redirect
  from bokeh.embed import server_document
  import secrets
  
  def embed_bokeh_app(request, app_name):
      # 登录校验
      if not request.user.is_authenticated:
          return redirect("/login/")
      # 应用访问权限校验
      if not request.user.app_permissions.filter(app_name=app_name).exists():
          return HttpResponseForbidden("您没有该应用的访问权限")
      # 生成绑定当前会话的认证令牌
      if "bokeh_auth_token" not in request.session:
          request.session["bokeh_auth_token"] = secrets.token_hex(16)
          request.session["bokeh_bound_user"] = request.user.id
      # 生成嵌入脚本，将认证参数传给Bokeh
      embed_script = server_document(
          url=f"http://127.0.0.1:5006/{app_name}",
          arguments={
              "auth_token": request.session["bokeh_auth_token"],
              "current_uid": request.user.id
          }
      )
      return render(request, "bokeh_page.html", {"script": embed_script})
  

  额外编写一个仅本地可访问的内部校验接口，供Bokeh侧验证令牌有效性，逻辑为校验传入的用户ID和令牌是否与Django会话中存储的一致，有效返回200，无效返回403即可。

• Bokeh侧逻辑对接
  每个Bokeh应用初始化时先做身份校验，校验通过后再加载数据、绑定交互逻辑，最小代码示例：

  from bokeh.io import curdoc
  from bokeh.models import Button, Div
  import requests
  import pandas as pd
  from sqlalchemy import create_engine
  
  # 初始化和Django共用的业务数据库连接
  db = create_engine("你的数据库连接串")
  
  # 从连接请求中获取认证参数
  req_args = curdoc().session_context.request.arguments
  token = req_args.get("auth_token", [b""])[0].decode()
  uid = req_args.get("current_uid", [b""])[0].decode()
  
  # 调用Django本地接口校验身份
  auth_resp = requests.get(
      "http://127.0.0.1:8000/internal/verify_bokeh_token/",
      params={"uid": uid, "token": token}
  )
  if auth_resp.status_code != 200:
      curdoc().add_root(Div(text="身份校验失败，请刷新页面重新登录"))
  else:
      # 加载全用户公共数据
      public_cost_df = pd.read_sql("SELECT * FROM public_cost_data", db)
      # 加载用户历史会话状态，实现持久化
      session_record = pd.read_sql(
          "SELECT state FROM user_session_state WHERE uid = %s ORDER BY update_time DESC LIMIT 1",
          db, params=(uid,)
      )
      current_session_state = session_record.iloc[0,0] if not session_record.empty else {}
  
      # 绑定保存结果的按钮逻辑
      def save_calc_result():
          # 替换为实际计算逻辑
          result = {"uid": uid, "calc_value": 12345}
          pd.DataFrame([result]).to_sql("user_calc_result", db, if_exists="append", index=False)
          # 同步更新会话状态
          pd.DataFrame([{"uid": uid, "state": str(current_session_state)}]).to_sql(
              "user_session_state", db, if_exists="append", index=False
          )
  
      save_btn = Button(label="保存运算结果", button_type="primary")
      save_btn.on_click(save_calc_result)
      curdoc().add_root(save_btn)
  

• 反向代理配置
  在Django路由配置或前置Nginx中添加规则，将所有/bokeh/*路径的请求（包括websocket升级请求）转发到本地127.0.0.1:5006端口，确保用户无法直接访问Bokeh服务。

需求对应说明 #

• 用户身份认证：所有请求先过Django登录校验，Bokeh建立连接时二次校验会话绑定的令牌，无有效身份无法加载应用
• 运算结果存库：Bokeh侧通过数据库连接，携带当前用户ID将结果写入对应用户表
• 公共数据读取：应用初始化时直接从公共数据表拉取全用户通用的成本类等数据
• 会话持久化存储：用户操作状态绑定用户ID存入数据库，下次登录加载应用时自动拉取恢复
• 多应用权限管控：Django侧维护用户-应用权限映射，嵌入脚本前先做权限校验，无权限用户无法获取对应应用的嵌入代码，也无法通过代理访问对应Bokeh应用路径

注意：Bokeh服务必须绑定本地回环地址，禁止直接对公网暴露，避免用户绕过Django的权限校验直接连接Bokeh。认证令牌和Django会话强绑定，不要存储在前端本地存储，降低令牌泄露风险。

内容的提问来源于stack exchange，提问作者Wiet