如何将Bokeh中TextInput的输入值传入MySQL查询语句？

我正在用Bokeh搭配Flask，结合自建MySQL音乐数据库制作仪表盘，需求是根据用户输入的参数生成SQL查询语句，将结果存入DataFrame后在图表展示。现在卡在如何把TextInput组件的输入值传入以下SQL查询：

query = f'SELECT * FROM `tracks_per_year` WHERE tracktitle LIKE "{text_input_value}"'
track_search_df = dbconnect.run_query_to_df(query, connection=cnx)

我觉得Bokeh官方文档这部分说明不够，只展示了创建TextInput组件和在前端打印输入内容的示例，没讲怎么把值传到后端用于查询。官方示例代码如下：

from bokeh.io import show
from bokeh.models import CustomJS, TextInput

text_input = TextInput(value="default", title="Label:")
text_input.js_on_change("value", CustomJS(code="""
    console.log('text_input: value=' + this.value, this.toString())
"""))

show(text_input)

根据Bokeh与Flask的集成方式，分两种场景实现输入值的获取与SQL查询对接：

方案1：Bokeh Server + Flask集成（后端Python回调） #

适合需要复杂后端逻辑的场景，直接通过Python代码获取输入值并处理查询，无需前端JS操作。

实现步骤&代码示例 #

from bokeh.models import TextInput, Button, ColumnDataSource
from bokeh.plotting import figure
from bokeh.layouts import column
from bokeh.server.server import Server
from flask import Flask
import pandas as pd
import mysql.connector

# 初始化Flask应用
flask_app = Flask(__name__)

# MySQL连接配置（建议用环境变量存储敏感信息）
db_config = {
    'user': '你的用户名',
    'password': '你的密码',
    'host': 'localhost',
    'database': '你的音乐数据库名'
}

def bokeh_dashboard(doc):
    # 初始化图表数据源
    source = ColumnDataSource(data=dict(year=[], track_count=[]))
    
    # 创建输入组件和查询按钮
    track_input = TextInput(title="搜索曲目:", value="")
    search_btn = Button(label="查询", button_type="primary")
    
    # 创建示例图表（根据你的数据结构调整）
    p = figure(x_axis_label='年份', y_axis_label='曲目数量', title='每年曲目统计')
    p.line('year', 'track_count', source=source)
    
    # 查询回调函数：获取输入值、执行SQL、更新图表
    def update_chart():
        track_title = track_input.value.strip()
        if not track_title:
            return
        
        # 参数化查询：避免SQL注入风险，绝对不能直接用f-string拼接用户输入
        query = 'SELECT * FROM `tracks_per_year` WHERE tracktitle LIKE %s'
        # 模糊查询需添加通配符
        query_param = f'%{track_title}%'
        
        # 执行查询并更新数据源
        cnx = mysql.connector.connect(**db_config)
        try:
            df = pd.read_sql(query, cnx, params=(query_param,))
            source.data = df.to_dict('list')
        finally:
            cnx.close()
    
    # 绑定按钮点击事件到回调函数
    search_btn.on_click(update_chart)
    
    # 组装仪表盘布局
    doc.add_root(column(track_input, search_btn, p))

# 启动集成了Flask的Bokeh Server
server = Server({'/': bokeh_dashboard}, app=flask_app, port=5000)
server.start()

if __name__ == '__main__':
    server.io_loop.add_callback(server.show, "/")
    server.io_loop.start()

方案2：纯Flask路由+前端AJAX请求（无Bokeh Server） #

适合轻量Flask应用，通过前端JS监听输入事件，发送请求到Flask接口获取数据后更新图表。

实现步骤&代码示例 #

Flask后端部分

from flask import Flask, request, jsonify
import pandas as pd
import mysql.connector

app = Flask(__name__)

db_config = {
    'user': '你的用户名',
    'password': '你的密码',
    'host': 'localhost',
    'database': '你的音乐数据库名'
}

# 查询API接口
@app.route('/search_tracks', methods=['POST'])
def search_tracks():
    data = request.get_json()
    track_title = data.get('track_title', '').strip()
    if not track_title:
        return jsonify([])
    
    # 参数化查询
    query = 'SELECT * FROM `tracks_per_year` WHERE tracktitle LIKE %s'
    query_param = f'%{track_title}%'
    
    cnx = mysql.connector.connect(**db_config)
    try:
        df = pd.read_sql(query, cnx, params=(query_param,))
        return jsonify(df.to_dict('records'))
    finally:
        cnx.close()

# 仪表盘页面路由
@app.route('/')
def index():
    from bokeh.models import TextInput, Button, ColumnDataSource, CustomJS
    from bokeh.plotting import figure
    from bokeh.layouts import column
    from bokeh.embed import components
    
    # 初始化数据源和图表
    source = ColumnDataSource(data=dict(year=[], track_count=[]))
    p = figure(x_axis_label='年份', y_axis_label='曲目数量', title='每年曲目统计')
    p.line('year', 'track_count', source=source)
    
    # 创建输入组件和按钮
    track_input = TextInput(title="搜索曲目:", value="")
    search_btn = Button(label="查询", button_type="primary")
    
    # 前端JS回调：发送AJAX请求并更新图表
    search_btn.js_on_click(CustomJS(args=dict(source=source, input=track_input), code="""
        const trackTitle = input.value.trim();
        if (!trackTitle) return;
        
        fetch('/search_tracks', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json'
            },
            body: JSON.stringify({ track_title: trackTitle })
        })
        .then(response => response.json())
        .then(data => {
            // 转换数据格式适配Bokeh数据源
            const years = data.map(item => item.year);
            const counts = data.map(item => item.track_count);
            source.data = { year: years, track_count: counts };
        })
        .catch(error => console.error('查询出错:', error));
    """))
    
    # 生成Bokeh组件的嵌入代码
    script, div = components(column(track_input, search_btn, p))
    return f"""
    <html>
        <head>
            <link rel="stylesheet" href="https://cdn.bokeh.org/bokeh/release/bokeh-3.3.4.min.css" />
            <script src="https://cdn.bokeh.org/bokeh/release/bokeh-3.3.4.min.js"></script>
        </head>
        <body>
            {div}
            {script}
        </body>
    </html>
    """

if __name__ == '__main__':
    app.run(debug=True)

关键注意事项 #

• 禁止直接用用户输入拼接SQL：必须使用参数化查询，否则会引发严重的SQL注入风险
• 根据项目规模选择方案：复杂后端逻辑优先用Bokeh Server集成模式，轻量应用用AJAX模式更灵活
• 建议使用数据库连接池，减少频繁创建/关闭连接的性能损耗

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