Using 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 

1. 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.

1. Set up the quick_publisher Django project:

1$ django-admin startproject quick_publisher

1. 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

 

1. Let’s now build a second Django application that handles posts:
   
   $ ./manage.py startapp publish
2. 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

 

1. 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.

 

1. This is when Celery enters the picture.
   First, ensure that it is installed:

$ pip install Celery

 

1. 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].

 

1. Start the Redis server on a different console by doing the following:$ redis-server
2. 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.

 

1. Go Back to main/models.py, 
2. Import the send_verification_email function 
3. 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.

 

1. 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)

 

1. When we alter a model, we must always migrate the database:

$ ./manage.py makemigrations 

 

$ ./manage.py migrate

 

1. 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).