Meilleure façon d'associer une liste générée à une tâche dans celery

Question

Je suis à la recherche de conseils sur la meilleure façon de faire correspondre une liste générée par une tâche à une autre tâche dans celery.

Disons que j'ai une tâche appelée parse qui analyse un document PDF et produit une liste de pages. Chaque page doit ensuite être transmise individuellement à une autre tâche appelée feed . Tout cela doit être placé dans une tâche appelée process

Donc, une façon de faire est la suivante :

@celery.task
def process:
    pages = parse.s(path_to_pdf).get()

    feed.map(pages)

Bien sûr, ce n'est pas une bonne idée parce que j'appelle get() à l'intérieur d'une tâche.

De plus, cela est inefficace, puisque mon parse est enroulée autour d'une fonction de générateur et est capable de produire des pages, ce qui signifie que il devrait être possible de mettre en file d'attente la première page pour l'alimentation avant que la dernière page ait été cédée par l'analyseur syntaxique. .

Une autre possibilité est de faire ceci :

@celery.task
def process:
    for page in parse.s(path_to_pdf).get():
        feed.delay(page)

Cet exemple implique toujours d'appeler get() à l'intérieur d'une tâche cependant. De plus, cet exemple est une simplification excessive et j'ai vraiment besoin de faire certaines choses après que toutes les pages aient été alimentées (c'est-à-dire dans une tâche chord ).

Je cherche la manière la plus optimale de faire cela dans le céleri. J'apprécierais tout conseil.

Merci !

Answer 1

Il est probablement trop tard pour que cela vous soit utile, mais vous souhaitez probablement utiliser une chaîne de tâches :

@celery.task
def process():
    return chain(parse.s(), feed_map.s())

@celery.task
def feed_map(pages):
    return feed.map(pages)

si vous avez une tâche finale, par exemple final vous pourriez faire ça :

@celery.task
def feed_map(pages):
    return chord(feed.map.s(page) for page in pages, final.s)