I have a requirement that all my Celery tasks must be called with a specific keyword argument. I want to check and use the value of the keyword before my task is executed. For instance, suppose I have the following:
@shared_task
def my_task(*args, **kwargs):
foo = kwargs.get('bar') # -> I don't want to copy this to all my tasks
# Do stuff here
How can I create a new decorator called my_special_shared_task so that the below is equivalent to the above:
@my_special_shared_task
def my_task(*args, **kwargs):
# Do stuff here
What about task inheritance? something like:
class BaseTask(celery.Task):
foo = "some_value"
@app.task(base=BaseTask)
def my_task(*args, **kwargs):
# Do stuff here
print(self.foo)
here is the documentation.