In many situations, we need to run some tasks automatically in software. Like, sending an automatic email or doing something specific automatically after a definite interval. These are called scheduled tasks.

Today we will see how to implement scheduled tasks in Django. There are many ways to do it. Like using Celery or using python packages like django-cronjob, django-background-tasks, etc.

Here we will use django-background-tasks. It is very lightweight and easy to use.

So, what we will do? We will create an app in which there will be a key-value pairs. We will set the value of the key to ‘null’ if it is not updated within 5 minutes.

Let’s start. In the beginning, we will create an app named ‘keyvalue’ and register it in our project. Then we will create our model. Prior to that, we will install ‘django-background-tasks’ first.

Install from PyPI:

pip install django-background-tasks

Add to INSTALLED_APPS:

INSTALLED_APPS = [
    # ...
    'keyvalue',
    'background_task',
    # ...
]

As ‘background_task’ is also an app. We will have to migrate our database:

python manage.py migrate

After migrating, if you go to your database, you will see the ‘background task’ database with ‘Tasks’ & ‘Completed Tasks’ table inside.

background tasks app

Now will create a function that will create a ‘background_task’ each time a ‘KeyValue’ object is created or updated. We will write this function inside a separate python script. We will name it ‘tasks.py’

tasks.py

from django.utils import timezone
from background_task import background
from .models import KeyValue


@background(schedule=300)
def time_to_live(key):
    obj = KeyValue.objects.get(key=key)
    data = obj.timestamp
    current_time = timezone.now()
    try:
        if current_time > obj.timestamp:
            obj.value = 'null'
            obj.save()
    except:
        print("error")

This ‘time_to_live()’ function takes the ‘key’ of the ‘KeyValue’ object. This function is decorated with a ‘background’ decorator to register a task.

This will convert the ‘time_to_live()’ into a background task function. When we will call it from regular code it will actually create a Task object and store it in the database. The database then contains serialized information about which function actually needs to run later on.

This does place limits on the parameters that can be passed when calling the function – they must all be serializable as JSON. Hence why in the example above a ‘key’ is passed rather than a KeyValue object.

As we are going to implement this using REST APIs. We need to write our serializer before APIs.

serializers.py

from rest_framework import serializers
from .models import KeyValue

# converting to json and validation for passed data+
class KeyValueSerializer(serializers.ModelSerializer):
    class Meta:
        model = KeyValue
        fields = ['key', 'value', 'timestamp' ]

Now, We will create two APIs inside our ‘views.py’. Though we can create a separate module for writing ‘APIs’ we won’t make it longer now. One API for creating objects and another for retrieving, updating, or deleting an object. Inside these APIs, we will call our ‘time_to_live()’ function.

views.py

# Create your views here.

class addPairView(generics.CreateAPIView):
    lookup_url_kwarg = 'key'
    serializer_class = KeyValueSerializer

    def create(self, request):
        data = request.data
        serializer = KeyValueSerializer(data=data)
        key = data['key']
        if serializer.is_valid():
            serializer.save()
            time_to_live(key)
            return Response(serializer.data, status=201)
        return Response(serializer.errors, status=400)


class KeyValueRUDApi(generics.RetrieveUpdateDestroyAPIView):
    lookup_url_kwarg = 'key'
    serializer_class = KeyValueSerializer
    queryset = KeyValue.objects.all()

    def put(self, request, key):
        key = self.kwargs.get("key")
        instance = KeyValue.objects.get(key=key)
        TTL = datetime.timedelta(minutes=5)
        new_time = timezone.now() + TTL
        instance.timestamp = new_time
        instance.save()
        time_to_live(key)
        return super().put(request)

    def patch(self, request, key):
        key = self.kwargs.get("key")
        instance = KeyValue.objects.get(key=key)
        TTL = datetime.timedelta(minutes=5)
        new_time = timezone.now() + TTL
        instance.timestamp = new_time
        instance.save()
        time_to_live(key)
        return super().patch(request)

We have created a new ‘KeyValue’ object’ with the ‘key=200’. You can see in the table below.

With the creation of that object a task will also be created. You will see inside Tasks table.

As you know we have set the time to 300 seconds that means 5 minutes. The task will run automatically after 5 minutes. But we will have to start running the tasks from the queue first. For that, we will open another terminal and activate the virtual environment. Then, we will execute the following management command…

python manage.py process_tasks

This will simply poll the database queue every few seconds to see if there is a new task to run.

Now, let’s wait for five minutes. After that, if you refresh the page, you will see that there is no object in the ‘Tasks’ table. When a task executes, it becomes a ‘Completed Task’. So, we will find it on the ‘completed task’ table.

Our task has run perfectly. But it’s time to check if it has changed the value of the object with key=200. Let’s go to the ‘Key Value’ table and see the result:

result:

For more exciting articles on django visit here.

+ posts

Author | Python-Django Developer

+ posts

Full-stack Developer (Python | Django | React | React-Native | Angular | Vue)