Enregistrer la tâche basée sur la classe dans

Question

J'utilise Celery version 4.0.2. Par rapport aux versions précédentes de Celery, il semble que les tâches basées sur les classes ne soient pas enregistrées automatiquement (c'est-à-dire si vous avez configuré la découverte automatique).

Cependant, je ne parviens même pas à enregistrer manuellement une tâche basée sur une classe.

Selon le changement Céleri journal:

http://docs.celeryproject.org/en/latest/changelog.html#version-4-0-1

depuis la version 4.0.1, il devrait être possible d'enregistrer la tâche manuellement:

from celery import Celery, Task 
app = Celery() 

class CustomTask(Task): 

    def run(self): 
     return 'hello' 

app.register_task(CustomTask()) 

Mais cela ne semble pas fonctionner. Est-ce que quelqu'un sait comment y parvenir?

J'ai essayé quelques suggestions qui sont discutées (en dehors de l'intégration d'un chargeur de tâche personnalisée mentionnée dans https://github.com/celery/celery/issues/3744):

Register Celery Class-based Task

https://github.com/celery/celery/issues/3615

https://github.com/celery/celery/issues/3744

Answer 1

On y est presque! Vous devez appeler delay() sur la tâche que vous avez enregistrée.

Cela fonctionne:

from celery import Celery, Task 

app = Celery() 


class CustomTask(Task): 
    def run(self): 
     return 'hello' 


task = CustomTask() 
app.register_task(task) 

task.delay() 

Answer 2

Si vous avez besoin shared_task décorateur:

from celery import Task, shared_task 

class CustomTask(Task): 
    def process(self): 
     return 'hello' 

@shared_task(bind=True, base=CustomTask) 
def custom(self): 
    self.process() 

process est un nom personnalisé qui commence la tâche (décorateur remplace la méthode run)

bind=True se fixe fonction une instance de classe

base=CustomTask définit une classe de base pour une tâche