在 Python 中使用 celery 调用类的实例方法

Posted 2023-02-24

【中文标题】在 Python 中使用 celery 调用类的实例方法

【英文标题】：Calling instance method of class using celery in Python

【发布时间】：2019-01-17 18:39:39

【问题描述】：

在旧版本的Celery 中，根据http://docs.celeryproject.org/en/3.1/reference/celery.contrib.methods.html，可以将实例方法转换为 celery 任务，如下例所示

from celery.contrib.methods import task
class X(object):
    @task()
    def add(self, x, y):
        return x + y

我使用的是 Celery 4.1，默认情况下没有这样的功能。我怎样才能以某种简单的方式自己实现这个功能？

让我举例说明我的要求。

from abc import ABC, abstractmethod

AbstractService(ABC):

    def __init__(self, client_id, x, y):
        self.client_id = client_id
        self.x = x
        self.y = y

    @abstractmethod
    def preProcess(self):
        '''Some common pre processing will execute here'''

    @abstractmethod
    def process(self):
        '''Some common processing will execute here'''

    @abstractmethod
    def postProcess(self):
        '''Some common post processing will execute here'''

Client1Service(AbstractService):

    def __init__(self, x, y):
        super(__class__, self).__init__('client1_id', x, y)

    # I want to run this using celery
    def preProcess(self):
        super().preProcess()

    # I want to run this using celery
    def process(self):
        data = super().process()
        # apply client1 rules to data
        self.apply_rules(data)
        print('task done')

    # I want to run this using celery
    def postProcess(self):
        super().postProcess()  

    def appy_rules(self, data):
        '''Client 1 rules to apply'''
        # some logic

我想在 django 项目中使用 celery 运行 Client1Service 类的 preProcess、process 和 postProcess。

如果我得不到任何解决方案，那么我将不得不在一些外部 celery 任务中运行 preProcess、process 和 postProcess 的逻辑，这会有点混乱。 建议我对此要求进行一些设计。

【问题讨论】：

这能回答你的问题吗？ using class methods as celery tasks

【参考方案1】：

尝试使用：

from celery import shared_task
@shared_task(bind=True)
def yourfunction():
    dosth()

这里有一个很好的教程：http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

【讨论】：

在我的情况下，方法在类内。 @shared_task(bind=True) 实例方法不起作用。

我明白了 - 试试这个：***.com/questions/9250317/…，它会让你的课程扩展 celery.Task。