将参数传递给自定义芹菜任务

Question

我只是想知道将参数传递给自定义任务装饰器的正确方法是什么。例如，我发现我可以按如下方式对芹菜任务进行子类化：

class MyTask(celery.Task):
def __init__(self):
# some custom stuff here
super(MyTask, self).__init__()
def __call__(self, *args, **kwargs):
# do some custom stuff here
res = self.run(*args, **kwargs)
# do some more stuff here
return res

并按如下方式使用它：

@MyTask
def add(x, y):
return x + y

但我希望能够将参数传递给此任务，并使其根据参数(或等效地，基于它正在装饰的功能)以不同的方式运行。我能想到的有两种方法。一种是通过将额外的 kwarg 传递给芹菜的任务包装器并显式指定基数，例如

@celery.task(base=MyTask, foo="bar")
def add(x, y):
return x + y

我可以在我的自定义任务中以self.foo访问，但这对我来说有点像作弊。另一种方法是检查self.task，并根据其值改变行为，但这似乎也有些矫枉过正。理想情况下，我想将 kwarg 直接传递给自定义任务类，

@MyTask(foo="bar")
def add(x, y):
return x + y

但是当然，这会产生一个MyTask的实例 ，我们不想要，无论如何都不会起作用。

关于正确方法的任何建议？

Answer 1

您可以添加使用类成员而不是实例成员。因此，您将在MyTask中定义__init__之外的参数，如下所示。然后，可以将此类用作 celery 任务的基类，并将这些新类成员用作自定义任务的参数。

注意：很遗憾，您无法在__init__中传递它们，因为您需要在装饰时实例化MyTask。

class MyTask(celery.Task):
foo = "default"
def __init__(self):
# some custom stuff here
super(MyTask, self).__init__()
def __call__(self, *args, **kwargs):
# do some custom stuff here
print(self.foo)
res = self.run(*args, **kwargs)
# do some more stuff here
return res

然后，您可以使用：

@celery.task(base=MyTask, foo="bar")
def add(x, y):
return x + y