Celery：自定义基类/基于子类的任务未显示在 app.tasks 下答案

【问题标题】：Celery: Custom Base class/child class based task not showing up under app.tasksCelery：自定义基类/基于子类的任务未显示在 app.tasks 下
【发布时间】：2016-12-06 05:53:14
【问题描述】：

我正在尝试将一些 celery 任务创建为类，但遇到了一些困难。这些类是：

class BaseCeleryTask(app.Task):

def is_complete(self):
    """ default method for checking if celery task has completed. """
    # simply return result (since by default tasks return boolean indicating completion)

    try:
        return self.result
    except AttributeError:
        logger.error('Result not defined. Make sure task has run!')
        return False

class MacroReportTask(BaseCeleryTask):

def run(self, params):
    """ Override the default run method with signal factory run"""
    # hold on to the factory
    process = MacroCountryReport(params)
    self.result = process.run()
    return self.result

但是当我初始化应用程序并检查 app.tasks（或运行工作程序）时，应用程序的注册表中似乎没有上述这些任务。其他基于函数的任务 (using app.task() decorator) 似乎注册得很好。

我将上述任务运行为：

process = SignalFactoryTask()
process.delay(params)

Celery worker 错误并显示以下消息： Received unregistered task of type None.

我认为我遇到的问题是：如何像处理基于常规函数的任务一样将自定义类添加到任务注册表？

【问题讨论】：

标签： python task celery

【解决方案1】：

遇到了完全相同的问题，花了几个小时才找到解决方案，因为我 90% 确定这是一个错误。在您的课堂任务中，尝试以下操作

class BaseCeleryTask(app.Task):

    def __init__(self):
        self.name = "[modulename].BaseCeleryTask"

class MacroReportTask(app.Task):

    def __init__(self):
        self.name = "[modulename].MacroReportTask"

似乎向应用程序注册它仍然存在名称未自动配置的错误。让我知道这是否有效。

【讨论】：

感谢您的帮助！我必须测试，但会告诉你它是否有效！