Céleri tâches dynamiques/cachant l'implémentation de céleri derrière une interface

Question

J'essaie de comprendre comment implémenter mes travaux asynchrones avec Celery, sans les lier à l'implémentation de Celery.

Si j'ai une interface qui accepte des objets à programmer, comme appelables (ou un objet qui enveloppe une appelable):

ITaskManager(Interface): 
    def schedule(task): 
     #eventually run task 

Et je pourrais le mettre en œuvre avec le module de foulage:

ThreadingTaskManager(object) 
    def schedule(task): 
     Thread(task).start() # or similar 

Mais il semble que cela ne pourrait pas être fait avec du céleri, ai-je raison?

Answer 1

Peut-être une, mais assez laid, solution pourrait consister à définir une tâche de céleri qui charge dynamiquement l'objet de la tâche qui est passée comme argument:

@celery.task 
def taskrunner(taskname): 
    taskModule = __import__(taskname) 
    taskModule.run() 

CeleryTaskManager(object) 
    def schedule(task): 
     taskrunner.delay(task.__file__) 


from mytask import run 

CeleryTaskManager().schedule(run)