Afreen KhalfeUsing Celery With Django for Background Task Processing
Web applications often begin basic but may grow fairly complicated, and the majority of them rapidly outgrow the task of just responding to HTTP requests. When this occurs, one must distinguish between what must happen immediately (often within the HTTP request lifecycle) and what may happen later. Why is this the case? Because little things like these make a difference when your application becomes overburdened with traffic. A web application’s activities are characterized as crucial or request-time operations and background duties, which occur outside of request time.
Request-time operations can be performed in a single request/response cycle without fear of the operation timing out or the user having a negative experience. This tutorial’s major focus is on background tasks. The Producer-Consumer Architecture is the most commonly utilized programming paradigm in this circumstance. Generally, consumers obtain jobs from the queue in a first-in-first-out (FIFO) or priority-based manner. Consumers are also referred to as employees, and we shall use that phrase throughout because it is congruent with the terminology used by the technologies presented.
Setting up all the things
- Let’s install Django, assuming you’re already comfortable with Python package management and virtual environments:
1$ pip install Django
I’ve chosen to create another blogging software. The application’s emphasis will be on simplicity. A user may easily register an account and quickly make and submit a post to the site.
- Set up the quick_publisher Django project:
1$ django-admin startproject quick_publisher
- Start with getting the app:
1$ cd quick_publisher
2$ ./manage.py startapp main
When starting a new Django project, it is preferable to start with a primary application that has a bespoke user model, among other things. Frequently run across constraints with the basic Django User model. Having a customized User model allows us to be more flexible.
)
)
user.is_admin = True
user.save(using=self._db)
return user
class MyUser(AbstractBaseUser):
email = models.EmailField(
verbose_name=’email address’,
max_length=255,
unique=True,
)
first_name = models.CharField(verbose_name=’first name’, max_length=30, blank=True)
last_name = models.CharField(verbose_name=’first name’, max_length=30, blank=True)
is_active = models.BooleanField(default=True)
is_admin = models.BooleanField(default=False)
objects = UserAccountManager()
USERNAME_FIELD = ’email’
REQUIRED_FIELDS = [‘first_name’,’last_name’]
def __str__(self):
return self.email
def has_perm(self, perm, obj=None):
“Does the user have a specific permission?”
# Simplest possible answer: Yes, always
return True
def has_module_perms(self, app_label):
“Does the user have permissions to view the app `app_label`?”
# Simplest possible answer: Yes, always
return True
@property
def is_staff(self):
“Is the user a member of staff?”
# Simplest possible answer: All admins are staff
return self.is_admin
If you are not familiar with how custom user models work, make sure to check the Django Make sure to check out the Django documentation.
Now we must instruct Django to utilize this User model rather than the default one. Add this line to the quick_publisher/settings.py file:
1AUTH_USER_MODEL = ‘main.User’
In the quick_publisher/settings.py file, we must also add the main application to the INSTALLED APPS list.
INSTALLED_APPS = [
‘django.contrib.admin’,
‘django.contrib.auth’,
‘django.contrib.contenttypes’,
‘django.contrib.sessions’,
‘django.contrib.messages’,
‘django.contrib.staticfiles’,
‘main’,
]
- Generate the migrations
- Apply them
- Set up a superuser to access the Django admin panel:
$ ./manage.py makemigrations main
$ ./manage.py migrate
$ ./manage.py createsuperuser
- Let’s now build a second Django application that handles posts:
$ ./manage.py startapp publish - In publish/models.py, construct a basic Post model:
from django.db import models
from django.utils import timezone
from django.contrib.auth import get_user_model
class Post(models.Model):
author = models.ForeignKey(get_user_model())
created = models.DateTimeField(‘Created Date’, default=timezone.now)
title = models.CharField(‘Title’, max_length=200)
content = models.TextField(‘Content’)
slug = models.SlugField(‘Slug’)
def __str__(self):
return ‘”%s” by %s’ % (self.title, self.author)
Hooking up the Post model to the Django admin is done in the publish/admin.py file, as seen below.:
from django.contrib import admin
from .models import Post
@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
pass
Finally, add the publish application to the INSTALLED APPS list to connect it to our project.
INSTALLED_APPS = [
‘django.contrib.admin’,
‘django.contrib.auth’,
‘django.contrib.contenttypes’,
‘django.contrib.sessions’,
‘django.contrib.messages’,
‘django.contrib.staticfiles’,
‘main’,
‘publish’,
]
We can now start the server and navigate to https://localhost:8000/admin/ to generate our first articles, giving us something to play with:
1$ ./manage.py runserver
The next step is to create a way to view the published posts.
# publish/views.py
from django.http import Http404
from django.shortcuts import render
from .models import Post
def view_post(request, slug):
try:
post = Post.objects.get(slug=slug)
except Post.DoesNotExist:
raise Http404(“Poll does not exist”)
return render(request, ‘post.html’, context={‘post’: post})
Associate a new view with an URL in quick_publish/urls.py.
from django.contrib import admin
from django.urls import path,include
from publish.views import view_post
urlpatterns = [
path(‘admin/’, admin.site.urls),
path(‘<slug:slug>’,view_post,name=’view_post’),
]
Create the template that renders the post in: publish/templates/publish/post.html.
<!DOCTYPE html>
<html>
<head lang=”en”>
<meta charset=”UTF-8″>
<title></title>
</head>
<body>
<h1>{{ post.title }}</h1>
<p>{{ post.content }}</p>
<p>Published by {{ post.author.first_name }} on {{ post.created }}</p>
</body>
</html>
We can now move forward to http://localhost:8000/the-slug-of-the-post-you-created/ in the browser.
It’s not exactly a miracle of web design, but making good-looking posts is beyond the scope of this tutorial.
Sending Confirmation Emails
- Let’s provide the User model an is verified flag and a verification uuid:
# main/models.py
import uuid
class MyUser(AbstractBaseUser):
email = models.EmailField(verbose_name=’email address’,max_length=255,unique=True, )
first_name = models.CharField(verbose_name=’first name’, max_length=30, blank=True)
last_name = models.CharField(verbose_name=’first name’, max_length=30, blank=True)
is_verified = models.BooleanField(verbose_name = ‘verified’, default=False)
is_active = models.BooleanField(default=True)
is_admin = models.BooleanField(default=False)
verification_uuid = models.UUIDField(verbose_name =’Unique Verification UUID’, default=uuid.uuid4)
2. To add the User model to the admin use this occasion:
from django.contrib import admin
from .models import User
@admin.register(User)
class UserAdmin(admin.ModelAdmin):
Pass
3. Let us make the following changes to the database:
$ ./manage.py makemigrations
$ ./manage.py migrate
We’ll make a callback that will be fired once a User model is generated. This code will be included after the User model declaration in main/models.py.
from django.db.models import signals
from django.core.mail import send_mail
from django.urls import reverse
def user_post_save(sender, instance, signal, *args, **kwargs):
if not instance.is_verified:
# Send verification email
send_mail(
‘Verify your QuickPublisher account’,
‘Follow this link to verify your account: ‘
‘http://localhost:8000%s’ % reverse(‘verify’, kwargs={‘uuid’: str(instance.verification_uuid)}),
‘from@quickpublisher.dev’,
[instance.email],
fail_silently=False,
)
signals.post_save.connect(user_post_save, sender=User)
Django does not send emails on its own; it must be linked to an email provider. You may enter your Gmail credentials in quick publisher/settings.py for convenience, or you can add your preferred email service.
Gmail configuration looks like as below:
EMAIL_USE_TLS = True
EMAIL_HOST = ‘smtp.gmail.com’
EMAIL_HOST_USER = ‘<YOUR_GMAIL_USERNAME>@gmail.com’
EMAIL_HOST_PASSWORD = ‘<YOUR_GMAIL_PASSWORD>’
EMAIL_PORT = 587
To test things out, go to the admin panel and create a new user using a genuine email address that you can easily verify. If everything went properly, you should have received an email with a verification link.
Verification of the account:
from django.shortcuts import render,redirect
from django.http import Http404
from .models import MyUser
# Create your views here.
def home(request):
return render(request, ‘main/home.html’)
def verify(request, uuid):
try:
user = MyUser.objects.get(verification_uuid=uuid, is_verified=False)
except MyUser.DoesNotExist:
raise Http404(“User does not exist or is already verified”)
user.is_verified = True
user.save()
return redirect(‘home’)
- Hook the views up in quick_publish/urls.py.
# quick_publish/urls.py
from django.contrib import admin
from django.urls import path,include
from publish.views import view_post
from main.views import home,verify
urlpatterns = [
path(‘admin/’, admin.site.urls),
path(”,home, name = ‘home’),
path(‘<slug:slug>’,view_post),
path(‘verify/<uuid>’,verify, name =’verify’),
]
Remember to include a home.html file in main/templates/main/home.html. The home view will display it.
An email will be received with a valid verification URL if everything goes well.
You can see how the account has been confirmed if you follow the URL and then check in the admin.
Sending Emails Asynchronously
Here’s the issue with what we’ve done thus far. You may have observed that the process of creating a user is a little sluggish. This is due to Django sending the verification email during the request time.
It works like this: we transmit the user info to the Django application. The program constructs a User model before connecting to Gmail (or another service you selected). Django waits for the response before returning a response to our browser.
- This is when Celery enters the picture.
First, ensure that it is installed:
$ pip install Celery
- In our Django application, we must now establish a Celery application:
# quick_publish/celery.py
import os
from celery import Celery
os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘quick_publisher.settings’)
app = Celery(‘quick_publisher’)
app.config_from_object(‘django.conf:settings’)
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
Celery is a task list. It gets tasks from our Django app and executes them in the background. Celery must be combined with other services that serve as brokers.
Brokers act as a middleman between the web application and Celery while transmitting messages. We’ll be utilizing Redis in this lesson. Redis is simple to set up, and we can get started with it right away.
Install the Redis Python library, pip install redis, and the bundle for using Redis and Celery: pip install celery[redis].
- Start the Redis server on a different console by doing the following:$ redis-server
- Include the Celery/Redis configuration quick_publisher/settings.py:
# REDIS related settings
REDIS_HOST = ‘localhost’
REDIS_PORT = ‘6379’
BROKER_URL = ‘redis://’ + REDIS_HOST + ‘:’ + REDIS_PORT + ‘/0’
BROKER_TRANSPORT_OPTIONS = {‘visibility_timeout’: 3600}
CELERY_RESULT_BACKEND = ‘redis://’ + REDIS_HOST + ‘:’ + REDIS_PORT + ‘/0’
Anything that may be run in Celery must first be specified as a task. Here’s how you do it:
# main/tasks.py
import logging
from django.urls import reverse
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from quick_publisher.celery import app
@app.task
def send_verification_email(user_id):
UserModel = get_user_model()
try:
user = UserModel.objects.get(pk=user_id)
send_mail(
‘Verify your QuickPublisher account’,
‘Follow this link to verify your account: ‘
‘http://localhost:8000%s’ % reverse(‘verify’, kwargs={‘uuid’: str(user.verification_uuid)}),
‘from@quickpublisher.dev’,
[user.email],
fail_silently=False,
)
except UserModel.DoesNotExist:
logging.warning(“Tried to send verification email to non-existing user ‘%s'” % user_id)
Here we moved the sending verification email functionality into another file called tasks.py.
- Go Back to main/models.py,
- Import the send_verification_email function
- Run it after a new user is created.
The signal code turns into:
from django.db.models import signals
from main.tasks import send_verification_email
def user_post_save(sender, instance, signal, *args, **kwargs):
if not instance.is_verified:
# Send verification email
send_verification_email.delay(instance.pk)
signals.post_save.connect(user_post_save, sender=User)
Take note of how we use the task object’s.delay function. This indicates we send the job to Celery and don’t wait for the outcome. If we used to send verification emails (instance.pk), we would still send them to Celery, but we would have to wait for the task to complete, which is not what we want. Celery is a service that has to be started. Open a new terminal, make sure the relevant virtualenv is enabled, then browse to the project folder.
$ celery -A quick_publisher.celery worker –loglevel=debug –concurrency=4
It will starts with four Celery process workers and it will look something like this:
[2022-11-30 14:56:17,565: INFO/MainProcess] celery@vaati-Yoga-9-14ITL5 ready.
Yes, you may now go ahead and create another user. Take note of how there is no delay, and check the logs in the Celery console to see if the jobs are correctly done. This should seem as follows:
[2022-11-30 14:58:38,165: INFO/MainProcess] Task main.tasks.send_verification_email[4f8f8455-3a61-48d2-b02f-ad6786b362e1] received
[2022-11-30 14:58:42,228: INFO/ForkPoolWorker-4] Task main.tasks.send_verification_email[4f8f8455-3a61-48d2-b02f-ad6786b362e1] succeeded in 4.0618907359967125s: None
This is what we want to do in our app. We’ll track how many times each post has been seen and provide the author a daily report. Every day, we’ll run through all of the users, get their posts, and send an email with a table comprising the posts and view counts.
- Let’s modify the Post model to handle the view counts situation.
class Post(models.Model):
author = models.ForeignKey(get_user_model(),on_delete= models.CASCADE)
created = models.DateTimeField(‘Created Date’, default=timezone.now)
title = models.CharField(‘Title’, max_length=200)
content = models.TextField(‘Content’)
slug = models.SlugField(‘Slug’)
view_count = models.IntegerField(“View Count”, default=0)
def get_absolute_url(self):
return reverse(“home”, args=[str(self.id)])
def __str__(self):
return ‘”%s” by %s’ % (self.title, self.author)
- When we alter a model, we must always migrate the database:
$ ./manage.py makemigrations
$ ./manage.py migrate
- Modify the view post Django view to count views as well:
def view_post(request, slug):
try:
post = Post.objects.get(slug=slug)
except Post.DoesNotExist:
raise Http404(“Poll does not exist”)
post.view_count += 1
post.save()
return render(request, ‘publish/post.html’, context={‘post’: post})
The view count might be displayed in the template. Include this <p>Viewed {{ post.view_count }} times</p> inside the publisher/templates/post.html file. Do a few views on a post now and see how the counter rises.
Let us make a Celery task. We’ll put it in publish/tasks.py because it’s about posts:
from django.template import Template, Context
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from quick_publisher.celery import app
from publish.models import Post
REPORT_TEMPLATE = “””
Here’s how you did till now:
{% for post in posts %}
“{{ post.title }}”: viewed {{ post.view_count }} times |
{% endfor %}
“””
@app.task
def send_view_count_report():
for user in get_user_model().objects.all():
posts = Post.objects.filter(author=user)
if not posts:
continue
template = Template(REPORT_TEMPLATE)
send_mail(
‘Your QuickPublisher Activity’,
template.render(context=Context({‘posts’: posts})),
‘from@quickpublisher.dev’,
[user.email],
fail_silently=False,
)
Remember to restart the Celery process whenever you make modifications to the Celery tasks. Celery must locate and reload tasks. Before we create a periodic job, we should test it in the Django shell to ensure that everything works as expected:
$ ./manage.py shell
In [1]: from publish.tasks import send_view_count_report
In [2]: send_view_count_report.delay()
Note: In your mail, you will receive a nifty little report in your email.
Now create a periodic task. Open up quick_publisher/celery.py and register the periodic tasks:
# quick_publisher/celery.py
import os
from celery import Celery
from celery.schedules import crontab
os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘quick_publisher.settings’)
app = Celery(‘quick_publisher’)
app.config_from_object(‘django.conf:settings’)
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
app.conf.beat_schedule = {
‘send-report-every-single-minute’: {
‘task’: ‘publish.tasks.send_view_count_report’,
‘schedule’: crontab(), # change to `crontab(minute=0, hour=0)` if you want it to run daily at midnight
},
}
So far, we’ve set up a schedule that will execute the job publish.tasks.send view count report every minute, as specified by the crontab() notation. You may also define different Celery Crontab schedules.
Open a new terminal, choose the proper environment, and launch the Celery Beat service.
$ celery -A quick_publisher beat
The Beat service’s role is to schedule and push jobs into Celery. Take into note that the schedule causes the send view count report job to execute every minute, as configured. It is suitable for testing but not for use in a live web application.
Conclusions
It’s best to keep unreliable and time-consuming tasks outside of the request timeframe. Worker processes should do long-running activities in the background (or other paradigms). Background tasks can be utilized for a variety of tasks that are not vital to the application’s core operation. Celery can also perform recurring tasks with the help of the celery beat service. Tasks may be made more dependable by making them idempotent and retrying them (maybe using exponential backoff).
Add comment