A Custom middleware can validate each http request for the set of instructions given in the middleware, before sending back a response. Let us see how we can write a custom Middleware. As an example, we will be implementing a custom Authentication Middleware here.

Django’s out-of-the box Authentication middleware only authenticates requests if the project has made use of the Django provided “User” model or has extended the same. If we have otherwise implemented a model from scratch for the user data, we will have to write a custom middleware for authentication and authorization.

I am implementing this in my twitter clone project, where I have used a custom model for user data. The details of the project can be found in my previous blogpost: Building a Twitter Clone with Django REST Framework – I

First, we will create a file in TUser App. We will call it AuthMiddleware.py. To start off, there are a few basic rules to be followed while defining a Middleware are:

1. The custom middleware must accept a get_response argument while defining the __init__ method. This is called only once when the web server starts.
2. Define the __call__() method. This is called each time a request is made.
3. Include the path of the above file in settings.py – in the “Middleware” section

This is how a sample middleware is defined:

#AuthMiddleware.py
from django.contrib.auth.backends import BaseBackend

class AuthMiddleware(BaseBackend):
    def __init__(self, get_response):
        self.get_response = get_response
    def __call__(self, request):
        return self.get_response(request)
    def CustomMiddleware(self,requst):
        #Steps to perform
        return True

#settings.py
MIDDLEWARE = [
    ...
    #custom middlewares:
    'TUsers.AuthMiddleware.AuthMiddleware'
]

Here’s my version of Custom Authentication Middleware – AuthMiddleware.py

class AuthMiddleware(BaseBackend):
    def __init__(self, get_response):
        self.get_response = get_response
    def __call__(self, request):
        self.auth(request)
        return self.get_response(request)
    def auth(self,request):
        exclusion_list = ['/signup','/login']
        if request.path not in exclusion_list:
            try:
                urlstr = request.path
                user = urlstr.split('/')[1]
                token = request.session.get('authtoken').get('token')
                payload = jwt.decode(token,settings.AUTH_TOKEN)
                userObj = TUser.objects.get(username=user)
                if payload.get('username') == userObj.username:
                    return True
                else:
                    raise PermissionDenied
            except (jwt.ExpiredSignature, jwt.DecodeError, jwt.InvalidTokenError) as e:
                error = {'Error_code': status.HTTP_403_FORBIDDEN,
                                'Error_Message': "Token is Invalid/Expired"}
                logger.error(e)
                raise PermissionDenied(error)
            except Exception as e:
                error = {'Error_code': status.HTTP_403_FORBIDDEN,
                                'Error_Message': "Invalid User"}
                logger.error(e) 
                raise PermissionDenied(error) 
        else:
            return True

This middleware will automatically validates each and every request made. In this project, the validation is performed to check if the token that a user obtains while Login is valid. If yes, the API end-point performs its action and returns the intended response. If not, an exception is raised and the access to the API is forbidden.

The complete file is available on github:

https://github.com/shilpavijay/TwitterClone/blob/main/TUsers/AuthMiddleware.py

You can find the complete project code here:

https://github.com/shilpavijay/TwitterClone

Previous posts:

Building a Twitter Clone with Django REST Framework – I

Building a Twitter Clone with Django REST Framework – II

In this post, let’s put some of the APIs to use:

• Create a new user using the ‘signup’ API end point

• Login to the application

• Login with wrong credentials

• Get all the followers

• Creating a tweet:

• Get the user’s Paginated timeline

• Like a tweet:

• Search for a tweet:

Previous post: Building a Twitter Clone with Django REST Framework – I

User Authentication

We will perform Authentication using JWT or JSON Web Token. JWT is a secret token that is generated and sent to the browser when a user successfully logs in with his credentials. Subsequently, whenever the user accesses any link, the token is sent back to the server in the request header. The server checks for the validity of the token before granting access.

User ————-> Login Successful ——————————–> Server: Generates JWT

User <————- Returns JWT to browser ——————– Server

User ————-> Sends JWT in request header ————> Server validates JWT signature

User <————- Access Granted ——————————– Server sends Response

We will be using Python’s JWT Module – PyJWT here. I did try out Django Rest Framework’s JWT, but I found it hard to get it working with little/unclear documentation. Also, the process of token generation and encryption is better understandable using PyJWT. It is also very flexible for customization. To install:

pip install PyJWT

settings.py

#In settings.py
import uuid

#RANDOM AUTHENTICATION TOKEN GENERATOR
AUTH_TOKEN = uuid.uuid4().hex.upper()

Let’s write the views for Authentication:

import jwt
 
#set Expiration time for the token. Best practise is to set it less than 10 minutes
EXP_TIME = datetime.timedelta(minutes=5)

def GetToken(username):
    '''
    Purpose: Get Access token
    Input: 
    username (mandatory) <str> Account user 
    password (mandatory) <str> Password
    Output: Token that expires in 60 minutes
    '''
    try:
        user = TUser.objects.get(username=username)
        if user:
            try:
                payload = {'id':user.id,'username':user.username,'exp':datetime.datetime.utcnow()+EXP_TIME}
                token = {'token':jwt.encode(payload,settings.AUTH_TOKEN).decode('utf8')}
                                # jwt.encode({'exp': datetime.utcnow()}, 'secret')
                return token
            except Exception as e:
                error = {'Error_code': status.HTTP_400_BAD_REQUEST,
                        'Error_Message': "Error generating Auth Token"}
                logger.error(e)
                return Response(error, status=status.HTTP_403_FORBIDDEN)
        else:
            error = {'Error_code': status.HTTP_400_BAD_REQUEST,
                        'Error_Message': "Invalid Username or Password"}
            return Response(error, status=status.HTTP_403_FORBIDDEN)
    except Exception as e:
        error = {'Error_code': status.HTTP_400_BAD_REQUEST,
                        'Error_Message': "Internal Server Error"}
        logger.error(e) 
        return Response(error,status=status.HTTP_400_BAD_REQUEST) 

def Auth(request,username):
    '''
    Purpose: Login to the Application
    Input: 
    token (mandatory) <str> user token 
    Output: User object of the logged in user
    '''
    try:
        token = request.session.get('authtoken').get('token')
        payload = jwt.decode(token,settings.AUTH_TOKEN)
        user = TUser.objects.get(username=username)
        if payload.get('username') == user.username:
            serializer = TUserSerializer(user)
            return Response(serializer.data, status=status.HTTP_200_OK)
        else:
            error = {'Error_code': status.HTTP_403_FORBIDDEN,
                        'Error_Message': "Invalid User"}
            logger.error(error)
            return Response(error,status=status.HTTP_403_FORBIDDEN) 
    except (jwt.ExpiredSignature, jwt.DecodeError, jwt.InvalidTokenError) as e:
        error = {'Error_code': status.HTTP_403_FORBIDDEN,
                        'Error_Message': "Token is Invalid/Expired"}
        logger.error(e)
        return Response(error,status=status.HTTP_403_FORBIDDEN) 
    except Exception as e:
        error = {'Error_code': status.HTTP_403_FORBIDDEN,
                        'Error_Message': "Internal Server Error"}
        logger.error(e) 
        return Response(error,status=status.HTTP_403_FORBIDDEN) 

@api_view(['POST'])
def Login(request,username=None,password=None):
    '''
    Authenticate if username and password is correct. 
    Input
    Output: return User object or Error 
    '''
    username = request.query_params.get('username')
    password = request.query_params.get('password')
    try:
        user = TUser.objects.get(username=username)
        if user.password == password:
            token = GetToken(username)
            user.token = token['token']
            user.save()
            request.session['authtoken'] = token
            serializer = TUserSerializer(user)
            return Response(serializer.data, status=status.HTTP_200_OK)
        else:
            error = {'Error_code': status.HTTP_400_BAD_REQUEST,
                        'Error_Message': "Invalid Username or Password"}
            return Response(error,status=status.HTTP_400_BAD_REQUEST) 
    except Exception as e:
        error = {'Error_code': status.HTTP_400_BAD_REQUEST,
                        'Error_Message': "Invalid Username"}
        logger.error(e) 
        return Response(error,status=status.HTTP_400_BAD_REQUEST)

App Users – REST APIs

urls.py: https://github.com/shilpavijay/TwitterClone/blob/main/TUsers/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path('signup', views.AccountSignup),
    path('login', views.Login),
    path('<str:username>/account', views.AccountUpdate),
    path('token', views.GetToken),
    path('auth', views.Auth),
    path('<str:loggedin_user>/<str:user>/follow', views.FollowUser),
    path('<str:username>/followers', views.GetFollowers),
    path('<str:username>/following', views.GetFollowing),
    path('<str:username>/block', views.Block_user), 
    path('users',views.users), #debugging
]

serializer.py : https://github.com/shilpavijay/TwitterClone/blob/main/TUsers/serializer.py

from rest_framework import serializers
from TUsers.models import TUser

class TUserSerializer(serializers.ModelSerializer):
    class Meta:
        model = TUser
        fields = '__all__'

views.py : This file content is too large to be placed here and is far more readable on Github. Link: https://github.com/shilpavijay/TwitterClone/blob/main/TUsers/views.py

App Tweets – REST APIs

urls.py: https://github.com/shilpavijay/TwitterClone/blob/main/Tweets/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path('create', views.CreateTweet),
    path('<str:username>/', views.Timeline), 
    path('<int:tweet_id>/delete', views.DeleteTweet),
    path('<int:tweet_id>', views.ShowTweet), 
    path('<int:tweet_id>/reply', views.Reply),
    path('<int:tweet_id>/retweet', views.Retweet),
    path('<int:tweet_id>/like', views.Like),
    path('search', views.Search),
    path('all_tweets', views.all_tweets), #debugging
]

serializer.py: https://github.com/shilpavijay/TwitterClone/blob/main/Tweets/serializer.py

from rest_framework import serializers
from Tweets.models import TCtweets
from django import forms

class TCtweetsSerializer(serializers.ModelSerializer):
    class Meta:
        model = TCtweets
        fields = '__all__'

class TCValidator(forms.Form) :
    username = forms.CharField()
    tweet_text = forms.CharField()

class ReplyValidator(forms.Form):
    username = forms.CharField()
    reply_text = forms.CharField()   

class RetweetValidator(forms.Form):
    username = forms.CharField() 

views.py: https://github.com/shilpavijay/TwitterClone/blob/main/Tweets/views.py

The complete code for this project can be found in my GitHub repository: https://github.com/shilpavijay/TwitterClone

In the next post, we will use some of these API end-points and see how they work.

Next post: Building a Twitter Clone with Django REST Framework – III

In this blog-series, I will be sharing my experience building an App with features similar to Twitter, using Django REST.

• There will be a code snippet for each step. You can find the entire code on GitHub too: https://github.com/shilpavijay/TwitterClone
• The approach I have taken is TDD – Test Driven Development. There will be a test case for each functionality, followed by the development of that feature.

Setting up the Django Environment

conda create env --name django_env
conda activate django_env

For Django REST Framework:

pip install djangorestframework

# Add to settings.py:
INSTALLED_APPS = [
     ...
    'rest_framework',
]

Let’s begin the Django Project

#To start a Django Project:
django-admin startproject TwitterClone

#Start the Apps:
python manage.py startapp TUsers
python manage.py startapp Tweets

#Add the two Apps to settings.py
INSTALLED_APPS = [
     ...
    'Tweets',
    'TUsers',
]

Writing Test Cases:

• Create a tests folder inside the App as shown below.
• This has to be made into a package by creating a ‘__init__.py’.
• All the files in the package should begin with a test_<>
• We will be writing test cases that initially fail and then coding the functionality, with the final goal of passing the test cases.

test_users.py and test_tweets.py

from django.test import TestCase
from TUsers.models import TUser

class TUserTestCase(TestCase):
    def setUp(self):      
        TUser.objects.create(username="mark_cuban",
                             password="sharktank",
                             country="USA")
    def test_create(self):
        self.assertEqual("USA",
                      TUser.objects.get(username="mark_cuban").country)

from django.test import TestCase
from Tweets.models import TCtweets
from TUsers.models import TUser

class TweetsTestCase(TestCase):
    def setUp(self):
        user1 = TUser.objects.create(username="lori",password="sharktank",country="USA")
        TCtweets.objects.create(username=user1,tweet_text="Hi there!")
    def test_create_tweet(self):
        user1 = TUser.objects.get(username='lori')
        self.assertEqual("Hi there!",TCtweets.objects.get(username=user1).tweet_text)

Adding the models:

TUser/models.py and Tweets/models.py

from django.db import models
from datetime import datetime

get_cur_time = datetime.now().strftime('%m/%d/%Y %I:%M:%S %p')

class TUser(models.Model):
    username = models.CharField(max_length=100,unique=True)
    password = models.CharField(max_length=100)
    country = models.CharField(max_length=100,null=True)
    modified_time = models.CharField(max_length=50,default=get_cur_time)
    following = models.ManyToManyField('self',related_name='followers',symmetrical=False,blank=True)
    blocked = models.BooleanField(default=False)
    token = models.CharField(max_length=10000,null=True)

from django.db import models
from TUsers.models import TUser
from datetime import datetime

get_cur_time = datetime.now().strftime('%m/%d/%Y %I:%M:%S %p')

class TCtweets(models.Model):
    username = models.ForeignKey(TUser,on_delete=models.CASCADE)
    tweet_text = models.CharField(max_length=100,null=True)
    time = models.CharField(max_length=50,default=get_cur_time)
    retweet = models.ManyToManyField(TUser,related_name="retweeted_users")
    like = models.ManyToManyField(TUser,related_name="liked_users")
    reply = models.ManyToManyField("self")     
    comment = models.CharField(max_length=100,null=True)

Swagger

Let us set up Swagger for API documentation:

pip install django-rest-swagger

#Add to settings.py
INSTALLED_APPS = [
     ...
    'rest_framework_swagger',
]

#Add the following to urls.py
from rest_framework_swagger.views import get_swagger_view

schema_view = get_swagger_view(title='Twitter Clone API')

urlpatterns = [
    path('admin/', admin.site.urls),
    path('docs/',schema_view),
]

If you are using Django 3.x, (which is my case here) there are a few things to be done in order to make Swagger work:

• Enable CoreAPI Schema: Django Rest Framework 3.x deprecated the CoreAPI based schema generation and introduced the OpenAPI schema generation in its place. Currently to continue to use django-rest-swagger we need to re-enable the CoreAPI schema. To settings.py add:

REST_FRAMEWORK = 
{'DEFAULT_SCHEMA_CLASS': 'rest_framework.schemas.coreapi.AutoSchema' }

• Change the swagger index.file: {% load staticfiles %} has been deprecated in django 2.x and removed in 3.x. Modify the same in index.html file of swagger in the location: …/Anaconda\envs\dj\Lib\site-packages\rest_framework_swagger\templates\rest_framework_swagger

...
{% load static %}  #Remove {% load staticfiles %}
<!DOCTYPE html>
<html lang="en">
<head>
...
...

And finally, you are done. You should see the Swagger API page.

Test Case for APIs

Too test the login API, first let’s write the test case and then build the code for the same. Here’s a sample:

test_users.py

   def test_api_login(self):
        client = Client()
        response = client.post('/login/', 
                   {'username':'mark_cuban'},format='json')
                    self.assertEqual(200,response.status_code)

In the next blogpost, we will see how to build the REST APIs.

You can find the project code on GitHub: https://github.com/shilpavijay/TwitterClone

Welcome back, to Python! It’s time to discuss the cool new features shipped with Python 3.7. Here are some of my favorites:

• Data Classes
• Built-in breakpoint
• Typing module
• Importing Data files

Data Classes

What is it? A new decorator: @dataclass

What’s new? It eases writing special methods in the class, like __init__(), __repr__(), and __eq__(. They are added automatically.

Example:

from dataclasses import dataclass, field
@dataclass(order=True)
class Test:
…field1: str
…field2: str

The following does not need an implementation of __repr__() in the class:

>>> t = Test("abc","xyz")
>>> t
Test(field1='abc', field2='xyz')
>>> t.field1
'abc'
>>> t.field2
'xyz'

Doesn’t need an implementation of __eq__ to do this:


>>> t1=Test("abc","xyz")
>>> t==t1
True
>>> t2=Test("a","b")
>>> t==t2
False

Breakpoint

What is it?  A built-in pdb.

What’s new? Not a new feature but simplifies using the debugger. Eliminates the necessity to import pdb.

Example:
def divide(a,b):
…breakpoint()
…return a/b

>>> divide(2,3)
> (3)divide()
(Pdb)

Old way of importing pdb:

def divide(a,b):
….import pdb; pdb.set_trace()
….return a/b

Typing module

What is it?  Annotations and Type hinting

What’s new? Function Annotations intend to provide a standard way of associating metadata to function arguments and return values.

• The annotations module and typing module provide hints on arguments and return value of a function.
• Annotations earlier worked only with names available in the current scope. i.e. Forward referencing was not supported.
• Annotations are now evaluated when a module is imported.

Example:  

Creating the file py37anno.py:

from __future__ import annotations

class Try:
…def foo(name: str) -> ‘salutation’:
……print(f”Annotations Example for you {name}”)

importing the annotations and typing module:

>>> from py37anno import Try
>>> from __future__ import annotations
>>> Try.foo.__annotations__
{‘name’: ‘str’, ‘return’: “‘salutation'”}
>>> import typing
>>> typing.get_type_hints(Try.foo)
{‘name’: <class ‘str’>, ‘return’: ForwardRef(‘salutation’)}
>>> Try.foo(‘Shilpa’)
Annotations Example for you Shilpa

Importing Data Files

What is it? An optimized, easy and organized API for working with data files in a Project

What’s new? Eliminates the necessity of hard-coding data file-names. The importlib.resources module helps in locating, importing and reading from the data file.

Example:

A file lorem.txt exists in the data directory of the project which also contains the __init__.py file

>>> import os
>>> os.listdir(‘data’)                                     #output: files in the ‘data’ directory
[‘lorem.txt’, ‘__init__.py’, ‘__pycache__’]

>> from importlib import resources
>>> with resources.open_text(“data”,”lorem.txt”) as f:
….print(f.readlines())
….
[‘”Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.” n’]

A note on installing python 3.7:

Creating a new environment:

conda create -n py37 -c anaconda python=3.7

Upgrade in an existing python environment to 3.7 in Anaconda:

conda install -c anaconda python=3.7