Plotly Dash回调优化：仅日期触发SQL查询，其余用Pandas过滤

要实现仅日期变更时查询数据库，其他筛选直接用Pandas过滤，核心是用Dash的dcc.Store组件缓存从SQL拉取的原始数据，拆分两个独立回调：

步骤1：添加缓存组件到布局 #

先在Dash布局里加入dcc.Store，用来存储日期范围对应的全量数据：

import dash
from dash import dcc, html, Input, Output, callback
import pandas as pd
import sqlalchemy

# 初始化app
app = dash.Dash(__name__)

app.layout = html.Div([
    # 筛选组件
    dcc.DatePickerRange(
        id='date-range-picker',
        start_date_placeholder_text="选择开始日期",
        end_date_placeholder_text="选择结束日期"
    ),
    dcc.Dropdown(id='type-dropdown', options=[...]), # 替换为你的实际选项
    dcc.Dropdown(id='item-dropdown', options=[...]),
    dcc.Dropdown(id='item-group-dropdown', options=[...]),
    dcc.Dropdown(id='county-dropdown', options=[...], multi=True), # 地区支持多选
    # 缓存组件：存储日期范围对应的原始数据
    dcc.Store(id='raw-data-store', data=[], storage_type='memory'),
    # 表格组件
    dash.dash_table.DataTable(id='table')
])

步骤2：第一个回调：仅日期变更时查询SQL #

这个回调只监听日期选择器的变化，把查询到的原始数据存入dcc.Store，必须用参数化查询避免SQL注入风险：

@callback(
    Output('raw-data-store', 'data'),
    Input('date-range-picker', 'start_date'),
    Input('date-range-picker', 'end_date'),
    prevent_initial_call=False # 页面加载时自动执行初始查询
)
def fetch_raw_data(start_date, end_date):
    if not start_date or not end_date:
        return [] # 日期未选择时返回空数据
    
    # 用SQLAlchemy执行参数化查询
    engine = sqlalchemy.create_engine('你的数据库连接字符串')
    query = """
        SELECT * FROM t 
        WHERE date BETWEEN :start_date AND :end_date
    """
    df = pd.read_sql(query, engine, params={'start_date': start_date, 'end_date': end_date})
    return df.to_dict('records') # 转成字典格式存入Store

步骤3：第二个回调：基于缓存数据做Pandas过滤 #

这个回调监听所有筛选组件和缓存的原始数据，只做内存中的Pandas过滤，不访问数据库：

@callback(
    Output('table', 'data'),
    Input('raw-data-store', 'data'),
    Input('type-dropdown', 'value'),
    Input('item-dropdown', 'value'),
    Input('item-group-dropdown', 'value'),
    Input('county-dropdown', 'value'),
    prevent_initial_call=False
)
def filter_data(raw_data, selected_type, selected_item, selected_item_group, selected_counties):
    if not raw_data:
        return []
    
    # 把Store中的字典转回DataFrame
    df = pd.DataFrame(raw_data)
    
    # 依次应用筛选条件
    if selected_type:
        df = df[df['type'] == selected_type]
    if selected_item:
        df = df[df['item'] == selected_item]
    if selected_item_group:
        df = df[df['item-group'] == selected_item_group] # 匹配你的SQL表列名
    if selected_counties:
        df = df[df['county'].isin(selected_counties)] # 修正原代码中的列名错误
    
    return df.to_dict('records')

关键细节说明 #

1. 回调拆分逻辑：第一个回调只处理慢操作（SQL查询），第二个处理快操作（内存过滤），完全分离不同触发源的逻辑
2. SQL注入防护：绝对禁止用f-string拼接SQL语句，参数化查询是安全标准
3. 变量/列名修正：修正了原代码中selected_tpye、selected_time_group、counties等错误命名
4. 初始加载控制：prevent_initial_call=False确保页面加载时自动完成初始查询和过滤

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