所有任务的Python Celery单个基类实例

Question

我有一个任务.py包含Task的子类。 根据docs，每个任务只实例化一次基类。在

但这只适用于相同的任务方法。调用不同的任务将创建一个新实例。所以我不能通过使用create_session创建的get_sessions访问会话。我怎么可能只有一个实例在不同的任务之间共享？在

class AuthentificationTask(Task):
    connections = {}

    def login(self, user, password, server):

    if not user in self.connections:
        self.connections = {user: ServerConnection(verbose=True)}
    # from celery.contrib import rdb
    # rdb.set_trace()

    self.connections[user].login(user=user, password=password, server=server)


@task(bind=True, max_retries=1, queue='test', base=AuthentificationTask)
def create_session(self, user, password, server):
    self.login(user, password, server)


@task(bind=True, max_retries=1, queue='test', base=AuthentificationTask)
def get_sessions(self, user, password, server):
    return self.connections[user].sessions

Answer 1

通过将connections作为AuthentificationTask上的类变量，您就走对了。这使得它可以作为类本身的属性来使用（例如，作为AuthentificationTask.connections）。当您在login方法中引用self.connections时，我相信Python正在寻找一个实例变量connections，而不是同名的类变量。对于所需的行为，将self.connections（在login和{}）替换为AuthentificationTask.connections。在

Answer 2

为Celery应用程序设置task_cls参数，如下所示：

class AuthentificationTask(Task):

    def example(self):
        logger.info('AuthentificationTask.example() method was called')


@celery.task(bind=True)
def test_my_task(self):
    # call AuthentificationTask.example
    self.example()

app = celery.Celery(
   __name__,
   broker='redis://localhost:6379/0',
   task_cls=AuthentificationTask,
   # other args
)

在本例中，所有任务都将使用自定义类作为默认值。在

Answer 3

似乎这是我的网站上的一个问题，因为每次重新初始化self.connections。在

self.connections = {user: ServerConnection(verbose=True)}

在进一步的测试中，base只为所有（不同）任务实例化一次。感谢@Danila Ganchar提出了另一种方法。我试试看！在