So far our Celery tasks where fire-and-forget jobs. We put a task in a queue and go to the next activity. However, sometimes we need to know if a task succeeded before we can do more work. Let us take a look on how Celery tracks tasks.
This post is part of my journey to learn Python. You can find the other parts of this series here. You find the code for this post in my PythonFriday repository on GitHub.
Downsides of state tracking
While it is a valid requirement to know the state of all your tasks, keep in mind that this comes with its downsides. If Celery needs to track the state of your tasks, it needs to do more work and that will reduce the throughput. The faster your Celery system should be, the better it is not to track the state of your tasks.
Backend options
Celery needs a backend to store the state of your task if you want to track it. There are two main operation models for the result backend: RPC (like RabbitMQ/QPid) or a database. Both have its pros and cons and you should check the documentation to get the right one for your application.
RabbitMQ as a backend
We can use RabbitMQ as a backend by setting the backend parameter to the value ‘rpc://’:
|
1 |
app = Celery('tasks', backend='rpc://', broker='amqp://guest@localhost//') |
To have enough time to check the different behaviour, I extended the wait period of the task to 10 seconds and return a string with the order_id:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
import time from celery import Celery from celery.utils.log import get_task_logger logger = get_task_logger(__name__) app = Celery('tasks', backend='rpc://', broker='amqp://guest@localhost//') @app.task(ignore_result=False) def prepare(order_id): time.sleep(10) # simulate work logger.info(f"order #{order_id} prepared") return f"#{order_id}" if __name__ == '__main__': app.worker_main() |
Check the state of your tasks
We need to start RabbitMQ and run Celery with our changed configuration. Next, we can start the REPL by typing python and enter these commands:
|
1 2 3 4 5 6 7 |
>>> result = prepare.delay(10) >>> result.state 'PENDING' >>> result.state 'PENDING' >>> result.state 'SUCCESS' |
The task is ‘PENDING’ at first and then changes to ‘SUCCESS’ when Celery is done processing it.
Get the result of your tasks
As soon as the state changes to ‘SUCCESS’, we can ask for the result of the task:
|
1 2 3 4 5 6 7 8 |
>>> result = prepare.delay(11) >>> result.result >>> result.state 'PENDING' >>> result.state 'SUCCESS' >>> result.result '#11' |
A database as a backend
Celery uses SQLAlchemy to talk to a result database, what gives us a wide range of options when it comes to the supported database systems.
We can set a connection string for the backend and SQLAlchemy will take care of the rest:
|
1 |
app = Celery('tasks', backend='db+sqlite:///./celery.db', broker='amqp://guest@localhost//') |
You can get the newest version of SQLAlchemy with this command:
|
1 |
pip install -U SQLAlchemy |
You can now run the examples from above and you should get a similar output.
Next
If you need to know what happened to your task, you can use a backend and Celery will track your tasks. Next week we take a closer look at another interesting feature of Celery: periodic tasks.
1 thought on “Python Friday #127: Track the State of a Celery Task”