Recently, I've had to write several complicated Celery tasks. Unfortunately, when doing a complicated process standard task functions can become unwieldy to write, read and unit test. After looking into how Celery tasks actually work, I was able to find a more manageable way of writing these complex tasks.
It turns out the
task decorator that you're used to using is just an object factory for
Task objects that turns the decorated function into the
run method of on the
Task instance it creates. Skipping the decorator and extending the
Task class directly makes things a little more flexible.
from celery import Task class MyTask(Task): ignore_result = True def run(self, source, *args, **kwargs): self.source = source data = self.collect_data() self.generate_file(data) def generate_file(self, data): # do your file generation here ... def collect_data(self): # do your data collection here ... return data
In this case
run is the equivalent of the function task you're used to, but thanks to OOP you're free to break some of the more complex code into logic blocks,
generate_file, and access to instance attribute,
To call the task your just need to instantiate the it and call the desired method to trigger it.
task = MyTask() # Run on Celery worker now task.delay(123) # Run on Celery worker at a sepcfic time task.apply_async(args=, eta=datetime(2015, 6, 5, 12, 30, 22)) # Run task directly, No celery worker task.run(123) task(123)
Testing also now becomes easier as well since you can test each unit on it's own.
For most cases, your standard function-based classes are probably going to do the job. But for those extra complex cases, class-based might make things easier work with.