In the last two parts of this series, we have discussed project setup and created endpoints required for implementing authentication. In this post, we will create endpoints for managing lists and tasks in a list. Below is a list of endpoints that we will be creating today. You can always find the API doc here.

  • GET /lists/list
  • POST /lists/add
  • GET /tasks/list
  • POST /tasks/add

Data Comes First

Before creating the endpoint, we will create our required models and migrate. Its a good practise to start with models as data becomes basis for your endpoint most of the time.

We need at least two models, one for lists and one for tasks. we will also create one more model for list access permissions. So, lets start with creating our first model

Models.py

Model represents a table in the database, and all its attributes are the fields in that table. Lets create a file called models.py in ToDo folder and all the models for ToDo will be written in this file.

For TaskList model, we need three fields

  • id - Unique Auto increment field
  • name - Char field of length 50
  • description - A Text field

We need not write SQL queries to create this, Django ORM does that for you, we just have to create a class which extends django.db.models.model as below

from django.db import models
from django.conf import settings

class TaskList(models.Model):

    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=50)
    description = models.TextField()

Similarly for Task Model we need id, name and description fields, and we will have to create a foreign key to List Model. We can do that by using Django's models.ForeignKey() as shown below

class Task(models.Model):

    id = models.AutoField(primary_key=True)
    list = models.ForeignKey('List', on_delete=models.CASCADE)  # Foreign key relation to List
    name = models.CharField(max_length=50)
    done = models.BooleanField(default=False)
    description = models.TextField()

Finally the ListAccess model, which is a mapping between user and list along with his access level on that list. Here we will create a foreign key to User model along with List model. One more field is created for role. This model looks as below.

class ListAccess(models.Model):

    # Foreign key relation to user model
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)

    # Foreign key relation to List
    list = models.ForeignKey('TaskList', on_delete=models.CASCADE)

    # this field provides access level of a user on a list
    role = models.CharField(max_length=5)

Time to migrate

We need to migrate, in order to have our additions or changes to be reflected on the actual database. Before we can migrate we must add ToDo to installed apps section in settings.py. Then, you can run makemigrations as shown below. Your tables are not created yet, it just creates the migrations files and can throw errors if some thing is not right.

$ python manage.py makemigrations ToDo
Migrations for 'ToDo':
  ToDo/migrations/0001_initial.py
    - Create model List
    - Create model ListAccess
    - Create model Task

Next step is to run migrate, this is when your migrations are applied on the database.

$ python manage.py migrate ToDo
Operations to perform:
  Apply all migrations: ToDo
Running migrations:
  Applying contenttypes.0001_initial... OK
  Applying auth.0001_initial... OK
  Applying ToDo.0001_initial... OK

Now you can inspect your DB, and can see the new tables created.

Protecting your endpoints

Django Rest Framework provides a few authentication classes, but we will create our own to work with the token based login system we created. Lets create a class called ToDoTokenAuthentication which extends rest framework's BaseAuthentication in a new file called authentication.py. Rest framework requires any authentication class to override the authenticate method which must return a tuple of (user, auth) if authentication is successful else either None can be returned or an AuthenticationFailed exception can be raised.

Our Authentication scheme expects the token to sent in http authorization header with Bearer prefix. So we start by getting the token and storing to a variable, then we will use pyjwt modules decode method to validate the token with the secret key set in settings. If the token is valid, we can query the auth_user table with username and return the user, in all other cases we consider the authentication is failed. We also check the type of token as we need an access token for authentication and not refresh token.


from django.contrib.auth.models import User 
from django.utils.six import text_type
from rest_framework import authentication, exceptions, HTTP_HEADER_ENCODING
import jwt
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned


class ToDoTokenAuthentication(authentication.BaseAuthentication):
    def authenticate(self, request):

        # Get the token from request header
        auth = request.META.get('HTTP_AUTHORIZATION', b'')
        if isinstance(auth, text_type):
            # Work around django test client oddness
            auth = auth.encode(HTTP_HEADER_ENCODING)

        auth = auth.split()

        if not auth or auth[0].lower() != 'bearer'.encode():
            return None

        if len(auth) == 1:
            msg = 'Invalid token header. No credentials provided.'
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = 'Invalid token header. Token string should not contain spaces.'
            raise exceptions.AuthenticationFailed(msg)

        try:
            token = auth[1].decode()
        except UnicodeError:
            msg = 'Invalid token header. Token string should not contain invalid characters.'
            raise exceptions.AuthenticationFailed(msg)

        # Validating the token
        try:
            decoded_token_payload = jwt.decode(token, settings.SECRET_KEY, algorithms='HS256')
        except jwt.exceptions.InvalidSignatureError:
            raise exceptions.AuthenticationFailed("Invalid Signature, Token tampered!")
        except jwt.exceptions.ExpiredSignatureError:
            raise exceptions.AuthenticationFailed("Token expired")
        except (jwt.exceptions.InvalidTokenError, jwt.exceptions.DecodeError):
            raise exceptions.AuthenticationFailed("Invalid Token")

        # Checking token type
        if not decoded_token_payload['type'] or decoded_token_payload['type'] != 'access':
            return None

        user = None
        try:
            global user
            user = User.objects.get(username=decoded_token_payload['username'])
        except ObjectDoesNotExist:
            raise exceptions.AuthenticationFailed("User doesn't exist")
        except MultipleObjectsReturned:
            raise exceptions.AuthenticationFailed("Multiple users found")

        return user, None



POST /lists/add

This endpoint should take name and description as body parameters and return a success message along with list object created. Lets start with creating a view and adding a route to it in urls.py.

This endpoint can be broken down to five steps:

  • Authenticating the user
  • Reading the request body parameters, here name is a required parameter and description is optional. So if name is not provided or is blank we will respond back with a error message
  • Creating a TaskList model object with request parameters and save it to database
  • Creating a ListAccess model object with User from request and TaskList object created in previous step and save to database
  • Finally respond back with appropriate message and data

The below code organised in the same sequence of steps as described above along with some exception handling

class ListAdd(APIView):
    authentication_classes = (ToDoTokenAuthentication,)
    permission_classes = (IsAuthenticated,)

    def post(self, request):
        if request.data.get('name', None) and request.data.get('name') != '':
            # Getting request data
            name = request.data.get('name')
            description = request.data.get('description') if request.data.get('description', None) else ''

            # Writing to database
            try:
                new_list = TaskList(name=name, description=description)
                new_list.save()
                new_list_access = ListAccess(user=request.user, list=new_list, role='owner')
                new_list_access.save()

                # Responding back
                resp_dict = {
                    'status': 'success',
                    'message': 'List created successfully',
                    'data': {'id': new_list.id, 'name': new_list.name, 'description': new_list.description}
                }
                resp = Response()
                resp.status_code = 201
                resp.data = resp_dict
            except ValueError as val_err:
                # Responding back
                resp_dict = {
                    'status': 'failed',
                    'message': 'Something went wrong while writing to database, {0}'.format(val_err),
                    'data': {}
                }
                resp = Response()
                resp.status_code = 400
                resp.data = resp_dict
            except Exception as er:
                # Responding back
                resp_dict = {
                    'status': 'failed',
                    'message': 'Something unexpected happened!, {0}'.format(er),
                    'data': {}
                }
                resp = Response()
                resp.status_code = 400
                resp.data = resp_dict

        else:
            resp_dict = {
                'status': 'failed',
                'message': 'List name is required but not provided',
                'data': {}
            }
            resp = Response()
            resp.status_code = 400
            resp.data = resp_dict

        return resp

For Authentication, we have used the ToDoTokenAuthentication class that we created earlier. We just have to provide the authentication_classes and permission_classes as shown above and rest framework automatically authenticates the user using the class we provided and responds back with failure message if authentication is failed else it lets the rest of the code to be executed.

Reading data from the request object is simple, we leverage data and data.get methods provided on the request object and verify if the required value i.e. name is sent in the body and is not empty.

The next step is to create model objects and save them using the Django ORM. We can create model object by instantiating the model class with values, then calling the save method on this object saves it to the database. First we need to create a TaskList object and save it to database and then create a ListAccess object with the task list object and save it. Here, the order of saving these objects is important as a ForeignKey constraint is set.

Now we can respond back with a success message and newly created list data if all the above steps are successful else we will respond back with an appropriate error message.

GET /lists/list

This endpoint will be fairly simple. The only main steps in this endpoint will be

  • Authenticate the user
  • Fetch lists belonging to this user from databse
  • Respond back with list of Lists

Like other endpoints, we need to create a view and add a route in urls.py.

Authentication in this endpoint will be exactly same as explained above, so, I will not discuss this here.

Next step is to fecth lists from database and filter for lists that the current user has access to. Access information is stored in ListAccess table so first we will filter this table for current user which we will get from request and select the list only. Here we have used values_list('list'), this is used to get values of one or more columns in a tuple. Since the column list is a foreign_key field, we will get the list of list_ids. We use these list_ids to filter and fetch lists from the TaskList model. Here we have used values() function on the query set, this function is very similar to values_list used above but it returns a dictionary instead of tuple.

Now we have the array of lists, which we are assinging to data key of resp_dict and sending in Response. This resp_dict is just for aesthetic purpose. we could have directly sent the array of list in response as well.

class ListFetch(APIView):
    authentication_classes = (ToDoTokenAuthentication,)
    permission_classes = (IsAuthenticated,)

    def get(self, request):
        resp_dict = {
            'status': '',
            'message': '',
            'data': None
        }

        try:
            list_ids = ListAccess.objects.values_list('list').filter(user=request.user)
            lists = TaskList.objects.filter(id__in=list_ids).values()
            resp_dict['status'] = 'Success'
            resp_dict['message'] = 'Retrieved the list of todo lists'
            resp_dict['data'] = lists

        except Exception as e:
            print(e)
            resp_dict['status'] = 'Failed'
            resp_dict['message'] = 'Something went wrong while fetching data. Error: '+e.__str__()
            resp_dict['data'] = None

        return Response(resp_dict)

POST /tasks/add

This endpoint will be very similar to POST /lists/add. One extra parameter we get in the request is the list_id to which this taks has to be added. We will have to check if a list exists with this id and the current user is the owner of this list before we make changes to the database, else we should respond with a permission error.

class TaskAdd(APIView):
    authentication_classes = (ToDoTokenAuthentication,)
    permission_classes = (IsAuthenticated,)

    def post(self, request):
        resp_dict = {
            'status': None,
            'message': None,
            'data': None
        }

        req_list_id = request.data.get("list_id")
        req_task_name = request.data.get("name")
        req_task_desc = request.data.get('description') if request.data.get('description', None) else ''

        if req_list_id and TaskList.objects.filter(id=req_list_id).exists() and \
                req_task_name and req_task_name != '':
            try:
                task_list = TaskList.objects.get(id=req_list_id)

                user_perm = ListAccess.objects.filter(user=request.user, list=task_list)

                if user_perm.count() != 1 or user_perm.first().role != 'owner':
                    raise PermissionError("You do not have permission to edit this list")

                new_task = Task(name=req_task_name, list=task_list, description=req_task_desc)
                new_task.save()

                resp_dict['status'] = "success"
                resp_dict['message'] = "Task creation successful"
                resp_dict['data'] = {"name": new_task.name, "description": new_task.description, "done": new_task.done,
                                     "list_id": new_task.list.id}
                resp = Response(resp_dict)
                resp.status_code = 200

            except PermissionError as pe:
                resp_dict['status'] = "failed"
                resp_dict['message'] = pe.__str__()
                resp_dict['data'] = None
                resp = Response(resp_dict)
                resp.status_code = 403
            except Exception as e:
                resp_dict['status'] = "failed"
                resp_dict['message'] = "Something went wrong, Error: "+e.__str__()
                resp_dict['data'] = None
                resp = Response(resp_dict)
                resp.status_code = 500

        else:
            resp_dict['status'] = "failed"
            resp_dict['message'] = "Invalid name or list_id passed"
            resp_dict['data'] = None
            resp = Response(resp_dict)
            resp.status_code = 400

        return resp

GET /tasks/list

This endpoint will take a single mandatory query parameter, list_id, and should return the list of tasks in that list. Similar to the previous endpoint, here we need to check if the user has access to this list before returning the data. Rest of the code is pretty straight forward.

class TaskFetch(APIView):
    authentication_classes = (ToDoTokenAuthentication,)
    permission_classes = (IsAuthenticated,)

    def get(self, request):

        resp_dict = {
            'status': None,
            'message': None,
            'data': None
        }

        try:
            list_id = request.query_params.get("list_id", None)

            # checking if the list id is provided
            if list_id is None or list_id == '':
                raise ValueError("Invalid list_id")

            # fetching list object
            try:
                task_list_obj = TaskList.objects.get(id=list_id)
            except ObjectDoesNotExist:
                raise ValueError("Invalid list_id")

            # checking if the user has permission on the given list
            try:
                list_perm_qs = ListAccess.objects.get(user=request.user, list=task_list_obj)
            except ObjectDoesNotExist:
                raise PermissionError("You do not have permission to access this list")

            # fetching tasks
            tasks = Task.objects.filter(list=task_list_obj).values()

            resp_dict['status'] = "success"
            resp_dict['message'] = "Fetched tasks successfully"
            resp_dict['data'] = tasks
            resp = Response(resp_dict)
            resp.status_code = 200

        except PermissionError as pe:
            resp_dict['status'] = "failed"
            resp_dict['message'] = pe.__str__()
            resp_dict['data'] = None
            resp = Response(resp_dict)
            resp.status_code = 403
        except ValueError as ve:
            resp_dict['status'] = "failed"
            resp_dict['message'] = ve.__str__()
            resp_dict['data'] = None
            resp = Response(resp_dict)
            resp.status_code = 400
        except Exception as e:
            resp_dict['status'] = "failed"
            resp_dict['message'] = "Something went wrong, Error: " + e.__str__()
            resp_dict['data'] = None
            resp = Response(resp_dict)
            resp.status_code = 500

        return resp

We will need one more endpoint to change the status of the task, I am not dicussing it here as it will be very similar to the previous endpoints.

With this, we have completed the creation of APIs. There are a couple of things I want to talk about before move on.

  1. Serializers
    In our endpoints where we have model objects and need to send it in the response as a JSON object we ended up creating a dictionary in place by accessing each value (you can find this in tasks/add endpoint). This is not a scalable approach, instead we shoud have created a serializer class and used that for the conversion part. Django rest framework also provides model serializers which help you achieve this conversion with ease. We haven't used serializers considering the length of this post.
  2. Test Cases
    It is always a good practice to write atleast the unit test cases for your code with a decent amount of coverage. We will try to dicuss this topic in another post.

In the next part we will dicuss the dockerization of this django application and how this can be deployed on AWS. You can find the code until this point in my github repository here. Please feel free to comment any question you have and do share your feedback.