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

Question

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

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 项目中使用芹菜运行preProcess、process和postProcessClient1Service类。

如果我得不到任何解决方案，那么我将不得不在一些外部芹菜任务中运行预处理、处理和后处理的逻辑，这将有点混乱。
建议我为此要求进行一些设计。

Answer 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