芹菜如何在for循环中创建任务组答案

【问题标题】：Celery how to create a task group in a for loop芹菜如何在for循环中创建任务组
【发布时间】：2021-09-04 16:27:05
【问题描述】：

我需要创建一个 celery 组任务，我想等待它完成，但我不清楚文档是如何实现的：

这是我目前的状态：

def import_media(request):
    keys = []
    for obj in s3_resource.Bucket(env.str('S3_BUCKET')).objects.all():
        if obj.key.endswith(('.m4v', '.mp4', '.m4a', '.mp3')):
            keys.append(obj.key)
    for key in keys:
        url = s3_client.generate_presigned_url(
            ClientMethod='get_object',
            Params={'Bucket': env.str('S3_BUCKET'), 'Key': key},
            ExpiresIn=86400,
        )
        if not Files.objects.filter(descriptor=strip_descriptor_url_scheme(url)).exists():
            extract_descriptor.apply_async(kwargs={"descriptor": str(url)})

    return None

现在我需要在组内为我拥有的每个 URL 创建一个新任务，我该怎么做？

我现在设法让我的流程像这样工作：

@require_http_methods(("GET"))
def import_media(request):
    keys = []
    urls = []
    for obj in s3_resource.Bucket(env.str('S3_BUCKET')).objects.all():
        if obj.key.endswith(('.m4v', '.mp4', '.m4a', '.mp3')):
            keys.append(obj.key)
    for key in keys:
        url = s3_client.generate_presigned_url(
            ClientMethod='get_object',
            Params={'Bucket': env.str('S3_BUCKET'), 'Key': key},
            ExpiresIn=86400,
        )
        if not Files.objects.filter(descriptor=strip_descriptor_url_scheme(url)).exists():
            new_file = Files.objects.create(descriptor=strip_descriptor_url_scheme(url))
            new_file.save()
            urls.append(url)
    workflow = (
        group([extract_descriptor.s(url) for url in urls]).delay()
    )
    workflow.get(timeout=None, interval=0.5)
    print("hello - Further processing here")
    return None

有什么优化的建议吗？至少现在它工作得很好！

提前致谢

【问题讨论】：

标签： python django celery django-celery

【解决方案1】：

https://docs.celeryproject.org/en/latest/userguide/canvas.html#groups

一个组运行所有任务，无论是否失败，如果前一个任务成功，则链接运行下一个任务。您可以做的不是每次在 for 循环中调用 apply_async 。您可以使用 signature 方法，该方法应用 args 但不执行任务，直到您准备好。

from celery import group
    ...
    all_urls= []
    for key in keys:
        url = s3_client.generate_presigned_url(
            ClientMethod='get_object',
            Params={'Bucket': env.str('S3_BUCKET'), 'Key': key},
            ExpiresIn=86400,
        )
        if not Files.objects.filter(descriptor=strip_descriptor_url_scheme(url)).exists():
            all_urls.append(url)
    g = group(extract_descriptor.s(kwargs={"descriptor": str(url)}) for url 
    in all_urls) # create group
    result = g() # you may need to call g.apply_sync(), but this executes all tasks in group
    result.ready()  # have all subtasks completed?
    result.successful() # were all subtasks successful?

【讨论】：