从CELERYBEAT_SCHEDULE内部将一个任务链接到另一个任务

Question

在 Django 项目中，在 Django 主应用程序目录中的tasks.py中定义了几个 Celery 任务。settings.py中的CELERYBEAT_SCHEDULE用于计划定期运行某些任务。

有没有办法将一个芹菜任务的输出链接到另一个芹菜任务，类似于下面？假设export_to_csv返回文件路径，upload_file接受文件路径作为输入。

CELERYBEAT_SCHEDULE = {
    'do-something': {
        'task': 'myproj.tasks.export_to_csv',
        'schedule': crontab(hour=12, minute=30),
        'args': (1, 2, 3),
        'options': {
            'link': 'myproj.tasks.upload_file'
        }
    },
}

类似于以下内容的工作：

from myproj.tasks import upload_file
CELERYBEAT_SCHEDULE = {
    'do-something': {
        'task': 'myproj.tasks.export_to_csv',
        'schedule': crontab(hour=12, minute=30),
        'args': (1, 2, 3),
        'options': {
            'link': upload_file.s()
        }
    },
}

但是由于循环进口，CELERYBEAT_SCHEDULE无法再位于settings.py。第一个示例不起作用，当"链接"字符串myproj.tasks.upload_file.s时也不起作用。

Answer 1

如果要在第二个函数执行之前等待一个函数完成，则根据定义，这是同步的。芹菜旨在异步执行函数。如果您要创建一个同步执行顺序相关函数的附加函数，并使用其中一个芹菜装饰器异步运行该函数，该怎么办？例如：

# tasks.py
def export_to_csv( ):
    # export code here
def upload_file( ):
    # upload code here
@shared_task
def export_to_csv_and_upload( ):
    export_to_csv( )
    upload_file( )
# settings.py
CELERYBEAT_SCHEDULE = {
    'do-something': {
        'task': 'myproj.tasks.export_to_csv_and_upload',
        'schedule': crontab(hour=12, minute=30),
        'args': (1, 2, 3),
        'options': {
            'link': 'myproj.tasks.upload_file'
        }
    },
}