python - programacion - utilizando métodos de clase como tareas de apio

poo python 3

Question

Intento utilizar los métodos de clase como las tareas de django-apio, marcándolo con @task decorator. La misma situación se describe here , preguntado por Anand Jeyahar. Es algo como esto

class A:
@task
def foo(self, bar):
...
def main():
a = A()
...
# what i need
a.foo.delay(bar) # executes as celery task
a.foo(bar) # executes locally

El problema es que incluso si uso una instancia de clase como esta a.foo.delay(bar) dice que ese foo necesita al menos dos argumentos, lo que hace que el self puntero falle.

Más información:

• No puedo convertir clase a módulo debido a la herencia
• Los métodos dependen en gran medida de los miembros de la clase, por lo que no puedo hacerlos estáticos
• Marcar la clase como la tarea con @task decorator hace que la clase sea una tarea en sí misma, y ​​podría ser posible ejecutar los métodos desde el método run() , usando algún argumento como clave para la selección del método, pero no es exactamente lo que quiero.
• Crear una instancia de clase y pasarla como self argumento a los métodos cambia la forma en que ejecuto los métodos, no como tomas de apio, sino como métodos habituales (es decir, mientras se prueban)
• Intenté averiguar cómo puedo registrar la tarea dinámicamente, desde el constructor, por ejemplo, pero el apio comparte el código entre los trabajadores, por eso es que parece imposible.

¡Gracias por tu ayuda!

Answer 1

Apio tiene soporte experimental para usar métodos como tareas desde la versión 3.0.

La documentación para esto está en celery.contrib.methods , y también menciona algunas advertencias que debes tener en cuenta:

http://docs.celeryproject.org/en/latest/reference/celery.contrib.methods.html

Tenga en cuenta : soporte para contrib.methods eliminado de Apio desde 4.0

Answer 2

Cuando tengas:

 a = A()

tu puedes hacer:

 A.foo.delay(a, param0, .., paramN)

Aclamaciones

Answer 3

Jeremy Satterfield tiene un tutorial claro y directo para escribir tareas basadas en clases si eso es lo que quiere lograr. Puedes consultarlo here .

La magia consiste básicamente en extender el celery.Task incluyen un método run() , como algo como esto:

from celery import Task
class CustomTask(Task):
ignore_result = True
def __init__(self, arg):
self.arg = arg
def run(self):
do_something_with_arg(self.arg)

y luego ejecuta la tarea de esta manera:

your_arg = 3
custom_task = CustomTask()
custom_task.delay(your_arg)

No estoy seguro si ignore_result = True parte ignore_result = True es necesaria o no por cierto.

Answer 4

Para mí, el único que funciona es apio.current_app porque esto solo pasa al método.

Entonces esto debería verse así:

from celery import current_app
from celery.contrib.methods import task_method
class A:
@current_app.task(filter=task_method, name=''A.foo'')
def foo(self, bar):
...

El nombre debe usarse si tiene un método con el mismo nombre en diferentes clases.