Передать аргумент пользовательской задаче сельдерея
Мне просто интересно, каким будет правильный способ передать аргумент декоратору пользовательской задачи. Например, я обнаружил, что могу создать подкласс задачи 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
но я хотел бы иметь возможность передать аргумент этой задаче и заставить ее вести себя по-разному в зависимости от аргумента (или, что то же самое, в зависимости от функции, которую она украшает). Я могу придумать два способа сделать это. Один из них заключается в передаче дополнительного kwarg в оболочку задачи celery и явном указании базы, например
@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, который нам не нужен и все равно не будет работать.
Любые предложения о правильном способе сделать это?
Ответы 1
Вы можете добавить использование члена класса вместо члена экземпляра. Следовательно, вы должны определить свои аргументы вне __init__ в 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