How to Use Asynchronous Tasks with Django and Celery?

Hi Dev,

This post will give you example of how to use asynchronous tasks with django and celery. if you want to see example of asynchronous tasks with django and celery then you are a right place. we will help you to give example of how to use celery in django. We will use django asynchronous tasks with celery.

When we are working with long running task our project now celery is very usefull to our task time to very quickly. like email sending, image processing and etc..

Celery workers are the foundation of Celery. Even if you intend to use Celery beat to schedule recurring tasks, a Celery worker will pick up your instructions and handle them at the scheduled time. Celery Beat adds a time-based scheduler for Celery workers to the mix.

Here i will give you we will help you to give example of how to use asynchronous tasks with django and celery. So let's see the bellow example:

In this step, we’ll create a new django project using the django-admin. Head back to your command-line interface and run the following command:

django-admin startproject example

Now we'll create a single app called core to store a list of post names. We're keeping things intentionally basic. Stop the local server with Control+c and use the startapp command to create this new app.

cd example
django-admin startapp core

Next, install the celery in our project.

python -m pip install celery

Install the redis in our local system:

sudo apt update
sudo apt install redis

redis-server
redis-cli

127.0.0.1:6379> ping
PONG
127.0.0.1:6379>

You sent the word ping to the Redis server after starting it with redis-cli, and it responded with an authoritative PONG. If you received this response, your Redis installation was a success.

In this step we need to require install python redis library following through below command.

python -m pip install redis

Next, create a celery.py file in example directory so open your file and paste below code.

import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_celery.settings")
app = Celery("django_celery")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
import core.tasks

Next, Update __init__.py file file in example directory so open your file and paste below code.

from .celery import app as celery_app

__all__ = ("celery_app",)

Then update INSTALLED_APPS within our settings.py file to notify Django about the app.

....
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'core',
]

CELERY_BROKER_URL = "redis://localhost:6379"
CELERY_RESULT_BACKEND = "redis://localhost:6379"

EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_USE_TLS = True
EMAIL_PORT = 587
EMAIL_HOST_USER = '[email protected]'
EMAIL_HOST_PASSWORD = 'YOUR_APP_PASSWORD'  

In this step We need to create a form that will be used .like add a bootstrap class and validation etc.. plus we need to add custom styling.

from time import sleep
from django.core.mail import send_mail
from django import forms
from core.tasks import send_contactus_email_task

class ContactForm(forms.Form):
    email = forms.EmailField(
        widget=forms.EmailInput(
            attrs={
                "placeholder": "Email",
                "class": "form-control"
            }
        ),error_messages={
               'required': 'The email field is required.'
        })
    message = forms.CharField(
        widget=forms.Textarea(
            attrs={
                "placeholder": "Message",
                "class": "form-control",
                "rows": 3
            }
        ),error_messages={
               'required': 'The message field is required.'
        })

    def send_email(self):
        """Sends an email when the contact form has been submitted."""
        # Simulate expensive operation(s) that freeze Django
        send_contactus_email_task.delay(
            self.cleaned_data["email"], self.cleaned_data["message"]
        )

In this step, we need to configure views. open the core/views.py file and add.

from core.forms import ContactForm
from django.views.generic.edit import FormView
from django.views.generic.base import TemplateView

class ContactFormView(FormView):
    template_name = "core/contact.html"
    form_class = ContactForm
    success_url = "/success/"

    def form_valid(self, form):
        form.send_email()
        return super().form_valid(form)

class SuccessView(TemplateView):
    template_name = "core/success.html"

In this step, we need to configure tasks. open the core/tasks.py file and add.

from time import sleep
from django.core.mail import send_mail
from celery import shared_task

@shared_task()
def send_contactus_email_task(email_address, message):
    """Sends an email when the contact form has been submitted."""
    # Simulate expensive operation(s) that freeze Django
    send_mail(
        "Contact Us",
        f"\t{message}\n\nThank you!",
        "[email protected]",
        [email_address],
        fail_silently=False,
    )

Next, then with your text editor create new templates files: core/templates/contact.html file and the add:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Tuts-Station.com</title>
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm>[email protected]/dist/css/bootstrap.min.css">
    <style type="text/css">
        body{
            background-color: #f7fcff;
        }
    </style>
</head>
<body>
    <div class="container mt-5 pt-5">
        <div class="row d-flex justify-content-center">
            <div class="col-md-8">
                <div class="card">
                    <div class="card-header">
                        <h5>How to Use Asynchronous Tasks with Django>and Celery? - <span class="text-primary">Tuts-Station.com</span></h5>
                    </div>
                    <div class="card-body">
                        <form method="post">
                            {% csrf_token %}
                            {{ form.as_p }}
                            <button type="submit" class="btn btn-success>>Submit</button>
                        </form>
                    </div>
                </div>
            </div>
        </div>
    </div>
</body>
</html>

<h1>Mail Send Successfully!</h1>

In this section, we need a urls.py file within the core app however Django doesn't create one for us with the startapp command. Create core/urls.py with your text editor and paste below code.

from django.urls import path
from . import views

urlpatterns = [
    path('', views.ContactFormView.as_view()),
    path('success/', views.SuccessView.as_view()),
]

Next, we require to add a URL path for our core app which can be done by importing include and setting a path for it.

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('core.urls')),
]

In this step, we’ll run the local development server for playing with our app without deploying it to the web.

python manage.py runserver

Open your second tab in your terminal without stop your django server..

python -m celery -A example worker -l info

Next, go to the http://localhost:8000/home address with a web browser.

So, in this example, the main goal is to sending email without taking any extra time to send email with backend celery.

I Hope It will help you....