python - 将参数传递给自定义 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

您可以添加使用类成员而不是实例成员。因此，您可以在 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