python - example - Apio: ¿Cómo ignorar el resultado de la tarea en cuerda o cadena?

django celery beat

Question

Estoy usando apio, tengo varias tareas que deben ejecutarse en orden.

Por ejemplo, tengo esta tarea:

@celery.task
def tprint(word):
    print word

Y quiero hacer algo como esto:

>>> chain(tprint.s(''a'') | tprint.s(''b''))()

Luego obtengo TypeError: tprint() takes exactly 1 argument (2 given) .

Lo mismo con acorde, en esta situación que necesito que se ejecute una tarea después de un grupo de tareas:

>>> chord([tprint.s(''a''), tprint.s(''b'')])(tprint.s(''c''))

Entonces, ¿cómo lidiar con esta situación? No me importa el resultado de cada tarea, pero deben ejecutarse en orden.

Agregar un segundo parámetro no funcionará:

@celery.task
def tprint(word, ignore=None):
    print word

>>> chain(tprint.s(''a'', 0) | tprint.s(''b''))()

Esto imprimirá ''a'' y ''None''.

Answer 1

Finalmente encuentre una solución para esto, un decorador de cadena hará este trabajo.

No sé cómo exactamente lo hizo el apio, pero el apio parece forzar el resultado de la tarea anterior al primer argumento de la próxima tarea.

Así que aquí hay un ejemplo:

def chain_deco(func):

    @functools.wraps(func)
    def wrapper(chain=None, *args, **kwargs):
        if chain is False:
            return False

        func(*args, **kwargs)
        return True

    return wrapper


@celery.task
@chain_deco
def hello(word):
    print "hello %s" % word

Ahora esto dará la salida correcta.

>>> (hello.s(word=''a'') | hello.s(word=''b''))()

O

>>> (hello.s(''a'') | hello.s(''b''))(True)

Y el decorador también proporciona la capacidad de detener una cadena en el medio (hacer que falle la cascada posterior).

El mismo mecanismo debería funcionar también para el chord .

Answer 2

Hay una funcionalidad incorporada para ignorar el resultado en el encadenamiento y otros - subtarea inmutable. Puede usar el acceso directo .si () en lugar de .s () o .subtask (inmutable = True)

Más detalles aquí: http://docs.celeryproject.org/en/master/userguide/canvas.html#immutability

Answer 3

Puedes intentar hacer algo como esto. En lugar de tener un solo parámetro para la función tprint, puede tener 2 parámetros

def tprint(word, x=None):
    print word

entonces

chain(tprint.s(''a'', 0) | tprint.s(''b''))()

Answer 4

Una posible solución ya ha sido publicada, pero me gustaría agregar más aclaraciones y una solución alternativa (y en algunos casos una superior).

El error que está viendo, que indica que la firma de su tarea necesita tener en cuenta un segundo parámetro, se debe al hecho de que al invocar tareas en una chain , apila automáticamente el result cada tarea como el primer parámetro de la siguiente tarea.

De los docs:

Las tareas se pueden vincular entre sí, lo que en la práctica significa agregar una tarea de devolución de llamada:

>>> res = add.apply_async((2, 2), link=mul.s(16))
>>> res.get()
4

La tarea vinculada se aplicará con el resultado de su tarea principal como primer argumento

Por lo tanto, en su caso, podría reescribir su tarea de esta manera:

@celery.task
def tprint(result, word):
    print word

Si no va a hacer nada con el resultado, también puede ignorarlo, cambiando el decorador de esta manera :

@celery.task(ignore_result=True)

Y luego no tendrá que cambiar la firma de su tarea.

Lo siento, ese último punto necesita más investigación.