将参数传递给自定义 celery 任务

Question

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

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

但我希望能够向此任务传递一个参数，并让它根据参数（或者等效地，根据它正在装饰的函数）表现出不同的行为。我可以想到两种方法来做到这一点。一种是向 celery 的任务包装器传递一个额外的 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

您可以添加使用类成员而不是实例成员。__init__因此，您可以在in之外定义您的参数，MyTask如下所示。然后，您可以使用此类作为 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