Asynchronous Tasks with Django and GCP Cloud Tasks

When building web applications, it is good practice to execute long-running tasks outside the HTTP request/response cycle.

For most Python web applications, Celery is the most commonly used tool for handling asynchronous task execution. However, for an application hosted on Google App Engine (GAE), Celery might be overkill, since GAE ships with cloud tasks; a fully managed service that allows you to manage the execution, dispatch and delivery of a large number of distributed tasks.

In this article, we are going to look at how to configure cloud tasks to handle asynchronous tasks in a Django app hosted on GAE.

Prerequisites

To follow along with this tutorial, you will need the following:

• A Google Cloud Platform Account
• A django app hosted on google app engine
• Google Cloud SDK Installed
• Billing enabled for your GCP account

Enable Cloud Tasks For Your Project

Let's start by enabling the cloud tasks API. From your GCP project page, navigate to the sidebar and look for cloud tasks under the tools section then enable the API for your project.

Create A Queue

Now that we have cloud tasks enabled, let's create a queue. You can think of a queue as a waiting area for tasks waiting to be executed. In your terminal, create a queue called "example queue" using the following gcloud command

gcloud tasks queues create example-queue

wait for the queue to initialize and then run the following command to verify that it was created succesfully

gcloud tasks queues describe example-queue

You should see an output similar to this:

name: projects/[project_name]/locations/[location]/queues/[QUEUE_ID]
purgeTime: '2020-12-04T10:07:54.059449Z'
rateLimits:
  maxBurstSize: 100
  maxConcurrentDispatches: 1000
  maxDispatchesPerSecond: 500.0
retryConfig:
  maxAttempts: 100
  maxBackoff: 3600s
  maxDoublings: 16
  minBackoff: 0.100s
stackdriverLoggingConfig:
  samplingRatio: 1.0
state: RUNNING

Include your queue configuration in your project's settings.py file

#settings.py


PROJECT_NAME = # your gcp project name
QUEUE_REGION = # your gcp region
QUEUE_ID = example_queue

Setup a task

In your app directory, create a new file called cloud_tasks.py

import json

from django.conf import settings

from google.cloud import tasks_v2beta3

client = tasks_v2beta3.CloudTasksClient()

def send_task(url, http_method='POST', payload=None):
    """ Send task to be executed """

    # construct the queue
    parent = client.queue_path(settings.PROJECT_NAME, 
             settings.QUEUE_REGION, queue=settings.QUEUE_ID)

    # construct the request body
    task = {
        'app_engine_http_request': {
            'http_method': http_method,
            'relative_uri': url
        }
    }

    if isinstance(payload, dict):
        # convert dict to JSON string
        payload = json.dumps(payload)

    if payload is not None:
        # The API expects a payload of type bytes
        converted_payload = payload.encode()

        # Add the payload to the request body
        task['app_engine_http_request']['body'] = 
        converted_payload

    # use the client to build and send the task
    response = client.create_task(parent=parent, task=task)

    return response

Here, we created a function that creates a task by making a post request to a URL endpoint that runs the task in the background.

in your views.py file and include the following

import time
from .cloud_tasks import send_task

def create_task(request):
    """ A simple view that triggers the task """
    task = "Example Task"
    send_task(url="/task/", payload=task)
    return JsonResponse({'message': "task created"})


def task_view(request):
    """ Processes a task """
    payload = request.body.decode('utf-8')
    time.sleep(60)
    print(f"{payload} is completed")

In the same directory, create a urls.py file

from django.urls import path

from .views import create_task, task_view

urlpatterns = [

path("", create_task, name="create_task"),

path("/task/", task_view, name="task_view")

]

Putting it all together

Deploy your application on GAE and navigate to the home route to trigger the create_task view. Next, navigate to the cloud tasks service page in GCP. You should see a queue (example-queue) with a task awaiting completion in it.

Conclusion

In this article, we looked at how to configure cloud tasks to handle long-running tasks for a Django application hosted on GAE.

Comments

[Horace FAYOMI]

Thanks for this, was very useful.