DEV Community

Samuel Adekoya
Samuel Adekoya

Posted on

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.


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

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.

Alt Text

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

You should see an output similar to this:

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

Include your queue configuration in your project's file

PROJECT_NAME = # your gcp project name
QUEUE_REGION = # your gcp region
QUEUE_ID = example_queue
Enter fullscreen mode Exit fullscreen mode

Setup a task

In your app directory, create a new file called

import json

from django.conf import settings

from 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'] = 

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

    return response
Enter fullscreen mode Exit fullscreen mode

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 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')
    print(f"{payload} is completed")
Enter fullscreen mode Exit fullscreen mode

In the same directory, create a 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")

Enter fullscreen mode Exit fullscreen mode

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.


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

Top comments (1)

fayomihorace profile image

Thanks for this, was very useful.