Professional Task Queues in Python with Celery, RabbitMQ & Redis

Key Takeaways

Today we're going to learn how to build professional task cues in Python by using celery rabbit MQ and we're also going to mix in some Reddis. Now these are central technologies, central tools that you need if you want to build complex applications. So I think you're going to learn a lot today. If you like the video, let me know by hitting a like button and subscribing to the channel. But now let us get right into it. [Music] Now before we jump right into the practical part, let's briefly cover what the individual tools are and how they relate to one another. Celery here is the central tool and the website says it's a simple, flexible and reliable distributed system to process vast amounts of messages. Basically with Celery we have two use cases. One is asynchronously processing tasks in cues and the other one is scheduling tasks on a regular basis. for example, every Wednesday at 2 p.m. or every 5 minutes. Additionally, it can distribute the work across multiple workers and machines, making it a distributed task Q. In this video today, we are going to touch on concurrency, but we're not actually going to use multiple machines. The interesting thing here is that celery basically communicates with itself. We have a celery client and a celery worker. In between, we have a message broker. We're going to talk about this in a second, but the idea is that the client submits a task that it wants to execute asynchronously to the message queue or to the message broker and the celery worker picks it up from there, pulls it out from there, processes it, and then stores the result in a backend, which is where Reddus comes in. We're going to talk about this in a second as well. So, what is the goal of the message broker? The main goal, especially in this scenario here, is to decouple the producer from the consumer. So, the client from the worker. client submits a task, produces a task and the worker consumes a task, processes a task. That is the basic idea. We want to decouple these two things so that they can work and communicate asynchronously. And what facilitates that is the message broker. Now, as the message broker, you can choose rabbit MQ, you can choose Reddis, you can also choose something else. The good thing about Rabbit MQ is that it also has mechanisms for guaranteeing delivery, for preventing message loss. So it's way more featurerich than something like Reddis. But you can also go with Reddis if you want to have a more minimal approach here. But at the end of the day, it's a matter of preference and a matter of what you need for your specific use case. As I said, Rabbit MQ is more featurerich. You can make a whole course about it. Reddus is more simple, straightforward, and doesn't have any mechanisms to guarantee that messages are resent in the case of a failure. And finally, there's the backend for salary, which is where the results end up. So if a salary task completes, sometimes it returns something. It doesn't have to return. You can also just take an action, produce a side effect, and then return nothing. This is also an option. But if you return something, it's going to end up in the back end. And the backend can literally be anything. It can be a Postgress database. It can be Reddius, it can be in memory, it can be Amazon S3. That's completely up to you. In our video today, in this video today, we're going to use Reddis in the first example or in the first couple of examples. And in a Django example, we're going to use the database that is associated with the Django application. In our case, that's going to be SQLite 3. But if you have your Django application set up to use Postgress, it's automatically going to go with Postgress. So to briefly summarize everything, the Celery client submits a task to the message broker. In our case, Rabbit MQ. This message broker handles buffering, queuing, reliable delivery. At some point the celery worker will pull out the task from the message queue process it and store the result if there is a result to be stored in the back end which can be reddis or whatever you set it up to be. That's it for the theory. Now let us move on with the practical part. All right. Now usually when you're using a text tech like this you're working on a more comprehensive application and you have a docker setup. So of course it's possible to install all of these components on your system. You can install Rabbit MQ, Reddis, Celery and run them natively on your system. But usually you're working on an application backend, front end, engine X and all this stuff and you have a Docker compose file and you just add these services in there. Since this is going to be the most likely way you work with these technologies. I'm going to do it like this. I'm not going to install the tools on my system. I'm going to write a Docker Compose file and then we're going to run all of this with Docker Compose. Now, if you don't know anything about Docker, I would recommend that you first check out a crash course on Docker. I have one on my channel. You can also watch someone else's crash course if you want to. But I think Docker is a prerequisite for understanding what we're doing here today. Not because it's a prerequisite for Celery, Rabbit and Q or Reddis, but because the whole setup here is going to be a Docker compos setup. So let us get started by creating a docker compose YAML file and defining our services. Now the first service that we define here is Rabbit MQ, our message broker. By the way, I'm going to copy paste these segments here to save time, but if this is too fast for you, you will find a link to my GitHub repository in a description down below. And there you will find all the files that we use in this video today. Now, for Rabbit MQ, we're using the image rabbit MQ3- management. And we're also setting two environment variables that will control the credentials for default user. I think by default, these are guest and guest. And by setting these environment variables, we change that to something that is custom. In my case here, I'm loading this from two other environment variables. These are going to be located in a file that we're going to create in a second called enth. There I'm going to put the credentials to have them in the central space. These are then automatically loaded from that file by docker compose. But you can of course also specify them here directly in the compose file. In addition to that, we define a volume in our case called rabbit mq data. This is to persist the data that rabbit MQ produces. For example, if the system crashes, but there's still a task in the queue, it's not going to be lost. It's persisted. It's on the disk. It's in the volume and can be used by any other container that mounts the same volume. Then we also map the port 15672. This is not the port for the actual communication, not the port for AMQP, the advanced message queuing protocol. This is for the management user interface. This is why we expose it. This is why we map it because we want to access it from the outside. The port that is used for the communication doesn't have to be mapped because it can stay in the network. The other services can access it anyway. And finally, we have a health check. This is important because other services are going to rely on rabbit MQ being started and being healthy, being active. And this is how we can check that this is actually the case. Next up, we have Reddis, which we're going to use as our results backend. Theoretically also possible to be used as a message broker. The image here is Reddis 7 Alpine. The command to start this is reddis server-safe61. This basically means if there's at least one change in the last 60 seconds write to disk. We also create a volume here reddis data to persist the reddis data and we also have a health check. Now the central part is the celery worker and the interesting thing is that the celery worker has the same build path or points to the same directory for building as the actual client. So the client and the worker are going to be in the same directory. This is the case now. This is also the case when we look at a Django application. Basically we say build current directory. This can also be backend for example. Then we have a command which says celery- a worker. So we have our worker application the command worker and this basically starts our celery worker. We say that we want to load the environment variables from which we still need to create. And then we have this depends on condition. Rabbit MQ and Reddis both have to be up and healthy. And the last service is our client. So this is the celery client. In our case, again, it's going to build current directory. If you had backend here, you would also put backend here or whatever you call this. The command here is just running a Python script that we don't have yet. And then it also loads the environment variables because they're going to be used there. And it depends on rabbit MQ and Reddis being healthy. and it depends on the celery worker at least having started. Now let us set up our environment variables. For this I'm going to create a file called and we're going to put here the rabbit mq credentials as well as the paths that celery is going to need. So for rabbit mq I'm just going to specify rabbit mq user rabbit mq pass and I'm going to call them custom user custom password. Quite simple in production of course use something safer. And for celery I'm going to define the broker and the backend URL. So the broker URL is the path to rabbit MQ in our case here AMQP advanced message queuing protocol custom user custom password. These are the credentials up here at rabbit MQ which we can use because we defined it in the docker compose file and then the port number 5672 which is the AMQP port number. For the back end we just specify reddis reddis again because we defined it in the docker compose 6379 because that's the port for reddis. So whatever you call these in the docker compose file, you can just use these names instead of localhost or IP addresses. So in this video, I want to show you multiple examples of how to use celery and rabbit MQ in Python. But we're going to start with a very very simple and trivial example. I'm going to define a file called worker py. And the only thing we're going to do is we're going to have a task here that randomly outputs a number with some delay. So these are going to be our imports. We're importing OS so we can access environment variables. We're importing time for the delay and random for the random number generation. And then of course we also import celery from celery. The first thing that we then do here is we define an application. So an instance of celery. In our case we call it random number. This is just the name of the application. Not too important. And then we specify the broker and the backend. These are the paths that we defined in our n file. So we say os.get salary broker URL and celery backend URL. So these are the values here. And now the only thing that we need to do to create a celery task is we need to annotate a function with the app.task decorator. In our case, the function is random number. It takes in some max value, waits for 5 seconds doing nothing, and then it returns random.rand int and a number between zero and the max value. So this is just simulating something that takes time. In reality, you would use something like an API call or you would process something on disk. You would do some calculations. something that takes time and would block the process. You do it as a task here so it can run in the background. In our case, it's just going to simulate something by waiting 5 seconds. This is basically the worker. That's all you have to do to create a simple celery task. Now, let us create the client. I'm going to call this client. py since this is the command that we use in our docker compos file. And here now we're going to submit a task and get the answer for this. The imports are the following. We're going to import time here as well in the client. You're going to see why in a second. Then we import from celery.result async result. This is a class. This allows us to create a result object that we can query. We can see if it's ready. We can retrieve the result if it is ready. So that is the class that we're going to use here in the client. And then from worker from our script that we just defined, we import the task so that we can actually submit it and also the application. Then what we do here is we wait 5 seconds in the beginning not for any functional reason just so we don't have all the cluttered output that is associated with booting all of this up with starting all the containers. So we're just going to wait 5 seconds and then we're going to submit a task by using the delay method. So what we do is we say result future this is not the result yet this is a promise this is a future result and what we do is we call method so random number delay we can do that because it's annotated with the apptask decorator from celery and we pass 100 which is the max value so let's go back to the worker max value is passed as a parameter this is a method but I can do method delay to submit this task using celery the result itself is going to be an asynchronous result. We pass here the ID of the future and the application that it's associated with and then we just print that the task was submitted. Now immediately after that we can also print the state of the asynchronous result. It's going to return pending because it's not yet processed. But what we can then do is we can say if this result is ready which is the same as saying if result.state is equal to success we can actually get the result and break out of the endless loop here. Otherwise, we can just print that the state is still pending or whatever it is. Wait a second and try again. Now, important. If you call result.get before it's ready, it's going to block. So, if I take this and I put it here, we're not going to see any of that because what's going to happen is it's going to say, okay, pending get. It's not done yet. So, it's going to wait here until the task is finished, until the task was executed successfully. Only then are we going to get the result and then we're going to continue with the rest of the code. If you only want to get the result once it's ready, you can do it like this. You can check okay not ready continue with the rest of the code. That is how you could do that. Now the final thing that we're missing before we can launch this is the docker file. So the docker file for python. This is as basic as it gets. We just take the python image. We set app as the work directory. We copy requirements txt. We install everything. We copy the data and we set the environment variable Python unbuffered to one. That's pretty standard actually. Finally, what we need is the requirements .txt file. Here we can put all the dependencies. And in our case, this is just going to be celery and then reddis in square brackets. Now all we have to do is we need to say docker compose up d-built. This will build a container from the image and then run it. So as you can see you get a lot of noise since Rabbit MQ is booting then the celery worker is booting and now we get the message submitted task pending pending pending pending and at some point we get 32 the random number has been generated. You can also see here the salary worker output that says task worker random number with some ID succeeded in a little bit more than 5 seconds result 32. And this is what the client receives and prints onto the screen. So let us now take a look at a more realistic and useful example. It's the same principle, the same concept but a different use case. What we do here is we again import OS. We import celery. In addition to that, we now also import open AI. You need to install this which is why we added OpenAI and Pantic here to the requirements txt file as well. And what we do here essentially is we're getting information about a movie in a structured way. So again we create an application. We call it movie info broker and backend stay the same. We create an open AI client. This is important. Now in my environment file, I also have now an open AI API key. So in the end file, you want to add a line with open AI API_key. Set it equal to your open AAI API key and then it's going to be automatically loaded here. I'm not going to show it because that's a secret key, but you can do the same thing and get access to open AAI. Then I define here a class movie. This is the information that I want to extract from a movie or for a movie. Title, release year, director, and genre. Quite simple. And then we're defining the actual task. In this case, the function is called movie info and takes in a prompt. This prompt is then sent to a large language model which responds with a structured output based on the movie class response format. So here we're no longer simulating waiting time. We actually have some waiting time because we're sending a request to an API and waiting for a response. In this case, it's going to be quite quickly answered, but in reality, if you have a large query, if you have a more complex model, it's going to take some time. I actually use this uh in a project right now, so I know what I'm talking about. And this is the worker. So, I can save that. And the client now interacts with it in the following way. We still import time, async, result, movie info, and app. We wait 5 seconds here again for the output. And this is basically the worker. Now, the client interacts with this worker in the same way as before. We have the same imports. We wait 5 seconds. And here I do three prompts in a row. So results future 1, 2, and three. I'm just asking for the movies Shutter Island, Inception, Predestination. I'm using the delay method for this. I'm taking these futures, adding them to a list, and then turning all of these elements into an async result with a list comprehension. Quite straightforward. And then I just iterate over these async results. And if one of them is ready, I get the answer, I print it, and then I remove it from the list. Quite simple. I do that with a waiting time of 1 second. Same idea as before, but now we have a useful task. So let's close this up and let's say docker compose up-built again. And you can see the three tasks were received. And I immediately get here title inception release year 2010. Christopher Nolan is the director, genre is science fiction. And then we have the same sort of answer here for predestination and shutter island. Now celery by default uses concurrency. So if you have multiple CPU cores available, it's going to use them. Which is why these three requests can be basically answered simultaneously. They don't have to be answered one by one. Now the pyantic model that we just used for movie was very basic and compact. In reality, however, you could be working with data models that are quite comprehensive and hard to fill in a single prompt. So maybe there's a limit, there's an actual hard limit, or you could just see that the performance decreases when you add too many fields. I myself am actually working on an application currently where I have to fill pretty large data models and to get reasonable results I have to split it up into multiple tasks. We can do exactly that with celery. So what we have here is a movie model split into three parts. We have movie part A, movie part B, movie part C, title, release year, director, genre, and a list of actors. Now in this case it doesn't make sense to split it up. The model is still quite basic and concise or compact. But in reality these could be 100 fields, 100 fields and 50 fields for example. So what we do here is we create a class for each of these parts and then we have also a task for filling each of these partial classes. We have movie info A, movie info B, movie info C which are basically the same function just with a different response format. And at the very bottom we also have a task for combining the parts. We pass parts here as a parameter. We have a merged empty dictionary and then we say here for part in parts merge.update part and then we return the merged dictionary. To now get a single result from celery for the client we can use groups and courts. So what we do first here is we define the prompt. We say that the header is equal to grouping these three functions. So we say movie info a b and c do s prompt. We group this together and then we say result is equal to chord of header combine parts s. So that is the final function that we call in the end and then we say combined is equal to result.get. What this results in is one output with all the data in a JSON object. So let's run this again by saying docker compose up-built. And as you can see what happens is we receive three different tasks. Then they return successfully and then we receive another task which is the combination task. And then as a result, the client prints title, Shutter Island, release year 2010, and so on and so forth with a list of the actors as well. Now, for the next example, we're going to get rid of the client, and we're going to add a new service called Celery Beat. Now, we're going to learn how to schedule stuff on a regular basis. For example, every 5 minutes, every 2 hours, or every day at 3:45 p.m., for example. Now, the command is almost the same. We're just using beat instead of worker. So, we're saying salary- a worker. This is the application and then the beat command, not the worker command. In addition to that, we also add a volume. This is not a proper docker volume. What we're doing here is we're mapping a directory called data in our current directory to the /data path in the container. And for this to work, we also need to do the same thing in the worker. Now, we can save that. And in the worker, we do all the work. So, we don't need the client anymore. We don't need anyone to submit a task. What we do here is we import from celery.scu. schedules the schedule and the cron tab function. The schedule function is used for executing a task every 10 seconds for example whereas the chronab allows us to specify times and dates. The task that we're using here is quite simple and straightforward. We have an output path which is /data/timestamp.txt. We create this directory and file if it doesn't exist. Otherwise, we just append to the file. And what we do is we append the current timestamp in ISO format. Now, as I said, there's two way to schedule stuff here. We can use schedule to say every 10 seconds. Or we can say schedule at a specific time every day. For example, 16:33 would be 4:33 p.m. In our case now, it's 16:46. So, let's go with 16:48. And I'm going to show you how this triggers. For this, I'm going to comment out this one so we don't confuse the two. Also, you can see I'm setting the time zone to be equal to Vienna. And I'm enabling UTC just to make sure that this is the same time that I have here on my system. So I can save this and now I can say docker compose up-build. Again, in the meantime, I'm going to open up a second terminal. Navigate to the same directory. And what we can see here is that I don't have it yet. But I now have the data directory. In here, I have nothing. But we can now watch this. I can say watch- N1 ls. And when a new file is created, we're going to see that it's listed here. There you go. You can see that task was successfully executed and now we have a timestamp.txt. So now I can stop this command and I can say cat timestamp txt and I see the timestamp that was written to this particular file. Now let's stop this. Let's go back into the worker and let's comment out this part. And now comment out this part. And let's say every 5 seconds I want to have something like this happening. Now I can do docker compose up and build again. I can also say watch- N1cat timestamp.txt. And now we should see every 5 seconds something added to this file. There you go. There you go again. And there you go again. Now, last but not least, I want to show you how all of this could look inside of a Django application. Now, since this would take too much time, I'm not going to show you everything step by step here, but you will find the code at the link in the description down below on my GitHub repository. And what you can do is you can just look at this project, open it in your editor and look for all the occurrences of note. So you can also do the same thing in the command line with rip grap. So rg note and it's going to show you all the notes that I've added here. I added these notes everywhere where I changed something, where I added something, where I um added a new file, changed something about the code, whatever. So if you want to see what's different compared to the default Django settings or the default docker compose that we used before, you can just look for note and you will find all the diffs. So the most important part is we have in our docker compose now uh the web service here. This is a new service that runs the actual Django application. Nothing too fancy. We still have the same uh rabbit MQ and uh celery. We don't have Reddus anymore because we're now using the Django DB back end which is basically SQLite in this case. And when we take a look at the code, we can go here to celery example. This is what I call this application. And you will see that I have a file called celery.py. What we do here is we get the settings from Django. We create a celery application movie project. This is the same LLM example that we had before. And we basically say auto discover tasks. So look for tasks and discover them. Run them. In the init py what I do is I say from celery import celery app celery. This is important so that this application can actually be found. And then in the settings, this is also very important. We define salary broker URL, salary result backend which is going to be part of the end file. We define the salary time zone and the salary enable UTC. Now this is important. I'm not going to show you the N file here, but what you need to do is you need to specify Django-d. So this is going to be Django DB as the actual backend. So instead of the reddest string, you just provide Django-d uh for the salary backend URL. You will also find the N file without all my keys in there in the GitHub example. And then basically in your application, you define the tasks in tasks. py. Here we use share task because we're using this in a reusable instance or in uh in a file that is part of the Django application. So in this case we don't say task we say share task and it basically picks whatever worker it has or is available right now. Then we define here the movie info function. We do the same structured output as before. And there's a couple of things that you need to change here and there. I'm not going to show absolutely everything here but the result of all this if you actually run this is going to be the following. Docker compose up-built. Now everything should be running on port 8000. If this is running correctly, you see all this output. Then it's running on port 8000. And you can say, tell me about Shutter Island. Submit. This is now going to show a task status page with a task ID processing pending. And when I reload, if the task is completed, it will show the result here. Very simple example. I'm not going to go through every single line of code here. But you can look at the code on the GitHub repository or in the GitHub repository. And if you are interested in a detailed Django Celery Rabbit MQ integration tutorial, let me know in the comment section down below. So that's it for today's video. I hope you enjoyed it and hope you learned something. If so, let me know by hitting a like button and leaving a comment in the comment section down below. Also, something that I usually don't really mention in my videos is I have a service page on my website. If you're interested in private tutoring or if you're interested in consulting or freelancing services, you might want to check that out. I just thought I could start mentioning this at the end of my videos. So, if you're interested in that, check out my page and specifically the services page. And besides that, don't forget to subscribe and hit the notification bell to not miss a single future video for free. Other than that, thank you much for watching. See you in the next video and bye.