How do I run code before task execution when using a shared_task in Celery

user_541729 0
Viewed 21

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
celery python
1 Answers

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.

user_10112530
Related