如何获取Celery完成任务的结果并在Flask中存储为变量？

核心问题分析 #

你当前的代码存在两个关键问题：

• Celery未配置结果后端，任务执行结果无法持久化，导致app.py无法获取
• app.py中错误调用两次tasks.scope，且传入的marketplace[0]是字符串首字符而非完整名称；同时直接调用任务是同步执行，既没利用Celery异步特性，也无法获取异步结果

步骤1：配置Celery结果后端并修复Selenium逻辑 #

修改tasks.py，添加结果后端配置（推荐和broker一致用Redis），同时修复WebDriver初始化问题（每个任务单独创建实例，避免并发冲突）：

from celery import Celery
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 配置消息队列和结果存储后端
celery = Celery(
    broker='redis://127.0.0.1:6379',
    result_backend='redis://127.0.0.1:6379'
)

@celery.task()
def scope(marketplace):
    # 每个任务单独初始化WebDriver
    web = webdriver.Chrome()  # 根据你的浏览器调整，如Firefox
    try:
        web.get(f'https://magiceden.io/marketplace/{marketplace}')
        # 获取价格文本
        price_check = WebDriverWait(web,30).until(
            EC.visibility_of_element_located((By.XPATH, "/html/body/div[2]/div[2]/div[3]/div[2]/div[2]/div[3]/div[2]/div[4]/div/div[2]/div[1]/div[2]/div/div[2]/div/div[2]/div/span/div[2]/div/span[1]"))
        ).text
        # 获取图片src（WebElement无法序列化，需提取可存储的字符串）
        image_src = WebDriverWait(web,30).until(
            EC.visibility_of_element_located((By.XPATH, "/html/body/div[2]/div[2]/div[3]/div[2]/div[2]/div[3]/div[2]/div[4]/div/div[2]/div[1]/div[2]/div/div[1]/div/div/img"))
        ).get_attribute('src')
        return (price_check, image_src)
    finally:
        # 确保任务结束后关闭浏览器
        web.quit()

步骤2：在app.py中获取任务结果 #

修改app.py，通过Celery的AsyncResult对象获取异步任务结果：

import tasks

marketplace = 'cheddar_block_games'

# 异步提交任务，返回任务状态对象
task = tasks.scope.delay(marketplace)

# 等待任务完成并获取结果（可设置超时，避免无限等待）
result = task.get(timeout=30)

# 解构结果赋值给目标变量
price_check, image_src = result

额外优化建议（针对Flask请求场景） #

如果是在Flask的HTTP请求中，直接用task.get()会阻塞请求20秒导致超时，建议改用异步视图+前端轮询的方式：

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/start-check')
def start_check():
    marketplace = 'cheddar_block_games'
    task = tasks.scope.delay(marketplace)
    # 返回任务ID，让前端轮询获取结果
    return jsonify({'task_id': task.id})

@app.route('/task-status/<task_id>')
def task_status(task_id):
    task = tasks.scope.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return jsonify({
            'state': task.state,
            'price_check': task.result[0],
            'image_src': task.result[1]
        })
    else:
        return jsonify({'state': task.state})

确保Redis服务正常运行，Celery Worker启动命令：celery -A tasks worker --loglevel=info

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