如何使用Celery实现调用Flask接口时自动调度该接口的执行

完全可以实现，不需要额外开发独立的触发函数，核心思路是将接口的业务逻辑和路由入口解耦，封装为可复用的Celery任务，路由收到请求时直接调度该任务执行即可。

具体实现步骤 #

• 将/get-products接口的核心业务逻辑封装为Celery异步任务，不要和路由定义耦合
• 路由函数收到请求后，直接调用任务的delay()方法提交执行，同步返回请求响应即可，全程不需要触发其他额外函数
• 如果有定期执行的需求，直接将这个Celery任务加入Celery Beat的定时调度列表即可，不需要再单独封装执行逻辑

代码示例 #

from flask import Flask
from celery import Celery
import time

app = Flask(__name__)
# Celery基础配置，可根据自身场景替换Broker、存储后端配置
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

# 接口核心逻辑封装为Celery任务
@celery.task
def get_products_core_logic():
    # 此处写入你原本/get-products接口的所有业务逻辑：数据库查询、数据加工、外部接口调用等
    time.sleep(10) # 模拟耗时业务操作
    products = [{"id": 1, "name": "示例产品1"}, {"id": 2, "name": "示例产品2"}]
    # 可按需将结果存入数据库、缓存，或者直接存在Celery结果后端后续查询
    return products

# /get-products接口定义
@app.route('/get-products')
def get_products():
    # 收到请求后直接调度任务执行，无额外函数调用
    get_products_core_logic.delay()
    # 同步返回响应，可根据业务需求返回任务ID、状态提示等内容
    return {"code": 200, "msg": "产品查询任务已提交执行", "data": None}

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

可选调整方案 #

• 如果需要接口同步返回执行结果，直接把get_products_core_logic.delay()替换为get_products_core_logic()同步执行即可，不需要修改其他逻辑
• 如果不需要异步执行，也可以直接在路由函数内部调用封装好的核心逻辑函数，不需要走Celery调度，按需切换即可

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