python - 如何在简单的 python Flask 应用程序中组织 celery 任务？

Question

我有一个使用 Celery 的简单 flask 应用程序。现在代码如下所示。

这是我的 server.py，位于根目录

from flask import Flask, request, jsonify
import os
import json
import logging
import requests
import sys

app = Flask(__name__)

# app.debug = True
app.config.update(
    CELERY_BROKER_URL=config.get('REDISTOGO_URL'),
    CELERY_RESULT_BACKEND=config.get('REDISTOGO_URL'),
)
celery = make_celery(app)

@celery.task(bind=False, default_retry_delay=30)
def convert(gif_url, webhook):
    // Do some heavy lifting task

@app.route("/convert", methods=["POST"])
def convert():
    // Call the task

这是我的make_celery(app)

from celery import Celery


def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

我感到困惑的原因是，如果我将任务提取到另一个名为 convert_task.py 的文件，那么该文件必须导入 app 和 make_celery 并在应用程序中必须导入 convert_task.py

我尝试过并且得到了

cannot import name 'app'

组织这个的Pythonic方式是什么？这背后的原因是我们将向 server.py 添加更多任务，我不希望它成为一个上帝文件。

Answer 1

我一直在开发一个使用 Celery 的 Flask 应用程序，所以这是我所做的对我有用的事情:

• make_celery 在创建应用实例的同一文件中定义(在您的情况下类似于 server.py)
• 我的任务在单独的 tasks.py 中定义，该任务与 server.py 位于同一目录中。这也是我实例化 celery 对象的地方(来自 make_celery)
• 启动 celery Worker 时，使用 -A 开关，然后为其提供tasks.py 的路径

我希望这对一些人有帮助，如果您对我如何组织我的应用程序有任何其他问题，请告诉我。