在Dash多页应用的回调中获取POST查询参数及Celery任务job值

核心思路 #

借助Dash基于Flask的底层特性获取请求参数，结合dcc.Store存储参数，再通过定时回调调用Celery的AsyncResult查询任务状态，进而更新进度条，完美适配你的多页应用和Docker部署环境。

1. 在Page One中获取请求参数并传递给回调 #

Dash基于Flask，可直接通过flask.request访问请求参数。将获取到的job参数存入dcc.Store（前端隐藏存储），供后台回调使用：

# page_one.py
from flask import request
import dash
from dash import dcc, html, Input, Output, State, callback

def layout():
    # 同时支持GET查询字符串和POST表单/JSON中的job参数
    job_id = request.args.get('job') 
    if not job_id and request.method == 'POST':
        # 处理POST请求的表单或JSON参数
        job_id = request.form.get('job') or request.get_json().get('job')
    return html.Div([
        dcc.Store(id='job-store', data=job_id),
        # 定时触发进度更新的组件，每秒一次
        dcc.Interval(id='progress-refresh', interval=1000, n_intervals=0),
        dcc.Progress(id='task-progress', value=0, max=100, style={'width': '100%'})
    ])

# 进度更新回调
@callback(
    Output('task-progress', 'value'),
    Input('progress-refresh', 'n_intervals'),
    State('job-store', 'data'),
    prevent_initial_call=False
)
def update_task_progress(n, job_id):
    if not job_id:
        return 0
    
    from celery.result import AsyncResult
    task_result = AsyncResult(job_id)
    
    # 根据Celery任务状态返回对应进度
    if task_result.state == 'PROGRESS':
        return task_result.info.get('progress', 0)
    elif task_result.state == 'SUCCESS':
        return 100
    elif task_result.state in ['FAILURE', 'REVOKED']:
        # 任务失败或取消时重置进度
        return 0
    # 任务未开始或处于其他状态时保持当前进度
    return dash.no_update

2. 配置Celery任务以支持进度更新 #

确保你的Celery任务在执行过程中主动更新状态，传递进度信息：

# celery_tasks.py（独立文件，供web和celery服务调用）
from celery import Celery

celery = Celery(
    'tasks',
    broker='redis://redis:6379/0',
    backend='redis://redis:6379/0'
)

@celery.task(bind=True)
def data_processing_pipeline(self, dataset_path):
    # 步骤1：数据集上传验证
    self.update_state(state='PROGRESS', meta={'progress': 20})
    
    # 步骤2：数据集划分
    self.update_state(state='PROGRESS', meta={'progress': 50})
    
    # 步骤3：模型训练
    self.update_state(state='PROGRESS', meta={'progress': 85})
    
    # 任务完成
    self.update_state(state='SUCCESS', meta={'progress': 100})
    return {'status': 'completed', 'model_path': '/models/trained_model.h5'}

3. 多页应用路由适配POST请求 #

默认Dash页面路由仅支持GET方法，若需要接收外部POST请求，需在app.py中自定义路由：

# app.py
import dash
from dash import html, dcc
from flask import request

app = Dash(__name__, use_pages=True)

# 为Page One注册支持POST的路由
@app.server.route('/page-one', methods=['GET', 'POST'])
def serve_page_one():
    return dash.page_registry['pages.page_one']['layout']()

app.layout = html.Div([
    html.Div([
        dcc.Link('Page One', href='/page-one'),
        dcc.Link('Page Two', href='/page-two'),
    ], style={'margin': '10px'}),
    dash.page_container
])

if __name__ == '__main__':
    app.run_server(host='0.0.0.0', debug=True)

4. Docker部署环境配置要点 #

• Redis/Celery连接：在web和celery服务的环境变量中统一设置：

  CELERY_BROKER_URL=redis://redis:6379/0
  CELERY_RESULT_BACKEND=redis://redis:6379/0
  

• PostgreSQL配置：若用Postgres存储应用数据，确保连接字符串正确：

  DATABASE_URL=postgresql://user:password@postgres:5432/your_db
  

• Nginx转发：确保Nginx允许POST请求并传递请求体：

  server {
      listen 80;
      location / {
          proxy_pass http://web:8050;
          proxy_set_header Host $host;
          proxy_set_header X-Real-IP $remote_addr;
          proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
          client_max_body_size 100M; # 适配数据集上传的大文件
      }
  }
  

业务场景适配 #

这套方案完全覆盖你的业务流程：

• 数据集上传后启动Celery任务，返回job_id
• 将job_id通过POST/GET请求传递到Page One
• 后台回调定时查询任务进度，更新进度条
• 任务完成/失败时自动更新进度状态

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