Building my first Flask API
Module 6 :
Welcome back,
This is our sixth and final module of building our flask API. In this module, we will add authentication to our API. Main idea here is Clients need to send us a JSON Web Token (JWT), when requesting one of our APIs. Only if a client’s JWT is valid we grant access to our API. JWTs are encoded JSON objects that carry information of a client in between requests. JWTs typically include informations such as, when a client session expires, what algorithm used to encode a token, and client’s unique ID.
In this module, we will discuss the following:
6.ADDING AUTHENTICATION TO OUR API
6.1 Creating a JWT Tokens
6.2 Adding Authentication for our GET route
6.3 Creating a Database for Users who are allowed API access
6.4 Creating Users ,who are allowed API access
6.5 Having User Login to get a Token
6.6 Creating Decorators to Add API Authentication
6.7 Adding Authentication for all our routes using Decorators
By the end of this module, you will be able to
- Create a user table in your database to store all the users who are granted access to our API.
- Use Flask’s SQLAlchemy extension as your ORM (Object-relational mapping) to simplify things.
- Use decorators to abstract all authentication code into one line of code to decorate a route.
Lets get started..
6.ADDING AUTHENTICATION TO OUR API
6.1 Creating a JWT(JSON WebToken)Tokens
JWT-JSON WebToken is a way to handle authentication of our API without having our client to send us a username and password over for every API call.
Sample JSON Web Token(JWT):
Sample JSON Web Token(JWT)
//Header
{
"alg": "HS256", #algorithm used to encode Token
"typ": "JWT"
}
//Payload
{
"id": 123456 #Client's unique id
"name": "Author RS"
"issued_at": 1516239022
"expiration_date": 1516939022 #when Client session expires
}
First thing first, import JWT and define a secret key in our app object. Define a route that allow someone to get a JWT token. For time being, we’re going to allow anybody to get this token, we’ll add actual authentication sooner. Define a route called login and within this define a method called get_token, which returns a token back to whoever calls this route. The jwt library has an encode method, make all the route to call the jwt encode method and return back a token to us. Call encode method, it takes in few parameters, the first parameter is the expiration date of this token, which is a dictionary object that has a key exp and the value an actual expiration_date . The second parameter is the actual SECRET_KEY that we have defined. The third parameter is the algorithm used for encoding, we set this to a default algorithm- HS256.
Define expiration_date. For that import the datetime library, we get the current time from this library, add a delta to that time. Here we add 100 seconds i.e., our token will last 100 seconds from the time client requested it.
In app.py add the following snippet
Code Snippet_6.1-Creating a JWT(JSON WebToken) Tokens
# In app.py add the following snippet:
import jwt, datetime #import jwt library which has JWT encode method
app.config['SECRET_KEY'] = 'spinach' #give any secretkey of ur choice
@app.route('/login')
def get_token():
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
TERMINAL:
Open up the terminal, start our server.From Postman fire a GET request. In the Response body you will get the token generated.
#TERMINAL:
Run server up
#POSTMAN:
#fire GET request at http://127.0.0.1:5000/login
#RESPONSE BODY:
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE1NzQ5MzA4MTN9.TCtCrIbGhzuCOZipvxN3ZLnNd9XhjKtRmh_Y_G8ChwI #<=token generated
6.2 Adding Authentication for our GET route
If anyone wants to view our book collection using our books route, then they need to pass us a valid token in order to view this resource. Instead of our client just passing in a GET request of books like before, now they have to send us in token using a query parameter. We have to parse the query parameters that are passed in. Run the jwt.decode method. The jwt.decode method throw an exception if token is invalid. For example, if the token that’s passed in to our client expired, or empty, or if the SECRET_KEY is wrong, then jwt.decode will throw an exception. Return a JSON response for error, and give a help string, say ‘Need a valid token to view this page’. If a valid token is passed, then return the jsonify of the books.
In app.py add the following snippet:
#In app.py import and add the following snippet:
import jwt, datetime #import jwt library which has JWT encode method
app.config['SECRET_KEY'] = 'spinach'
@app.route('/login')
def get_token():
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
In app.py edit decorators, as follows:
# In app.py edit decorators
#GET/books?token=asok9f09rtwi978hs
@app.route('/books')
def get_all_books():
token = request.args.get('token')
try:
jwt.decode(token, app.config['SECRET_KEY'])
except:
return jsonify({'error': 'Need a valid token'}),401
return jsonify({'Books':Book.get_all_books()}) #from database
Runnable Code Snippet_6.2-Adding Authentication for our GET route:
#RSapp.py
from flask import Flask, jsonify,request,Response
from BookModel import *
from settings import *
import json #import Json
from settings import * #import all
import jwt, datetime #import jwt library which has JWT encode method
app.config['SECRET_KEY'] = 'spinach'
@app.route('/login')
def get_token():
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
#request body
#{
# 'name' : 'Newbook',
# 'price' : 456,
# 'isbn' : 369852
#}
#Sanitize the data received from Client
#data sanitizer
def validBookObject(bookObject):
#if entry already exists
for i in books:
if i["isbn"] == bookObject["isbn"]:
return False
else:
#if "keyword" in dictionary
if ("name" in bookObject and "price" in bookObject and "isbn" in bookObject):
return True
else:
return False
#Adding a POST route
#POST/books
@app.route('/books', methods=['POST'])
def add_book():
request_data = request.get_json()#get the req.data by get_json method
if(validBookObject(request_data)): #considering only validObject is sent
Book.add_book(request_data['name'], request_data['price'], request_data['isbn'])
response = Response("", status=201, mimetype='application/json')
response.headers['Location'] = "/books" + str(request_data['isbn'])
return response
else:
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
#GET/books?token=a9f09rtwi978hs
@app.route('/books')
def get_all_books():
token = request.args.get('token')
try:
jwt.decode(token, app.config['SECRET_KEY'])
except:
return jsonify({'error': 'Need a valid token'}),401
return jsonify({'Books':Book.get_all_books()}) #from database
#GET/books/isbn
@app.route('/books/<int:isbn>')
def get_by_isbn(isbn):
return_value=Book.get_by_isbn(isbn)
return jsonify(return_value)
#PUT /books/5869
#{
# 'name': 'NewName',
# 'price': 456
#}
def valid_put_request_data(request_data):
if ("name" in request_data
and "price" in request_data):
return True
else:
return False
#PUT route
@app.route('/books/<int:isbn>', methods=['PUT'])
def replace_book(isbn):
request_data=request.get_json()
if(not valid_put_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
Book.replace_book(isbn, request_data['name'],request_data['price'])
response = Response("", status=204) #No content created
return response
# PATCH /books/isbn
#{
# 'name': 'UpdateNameAlone'
#}
# PATCH /books/isbn
#{
# 'price': 'UpdatePriceAlone'
#}
def valid_patch_request_data(request_data):
if ("name" in request_data
or "price" in request_data):
return True
else:
return False
@app.route('/books/<int:isbn>', methods=['PATCH'])
def update_book(isbn):
request_data = request.get_json() #we get the JSON data
if(not valid_patch_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error": "Invalid book object passed in request",
"helpstring": "Data should be in following format{'name': 'bookname', 'price':15}"
}
response = Response (json.dumps(invalidBookObjectErrorMsg),staus=404,mimetype='application/json')
return response
if("name" in request_data):
Book.update_book_name(isbn, request_data['name'])
if("price" in request_data):
Book.update_book_price(isbn, request_data['price'])
response = Response("",status=204) #204=Success
response.headers['Location'] = "/books/" + str(isbn)
return response
#DELETE /books/isbn
@app.route('/books/<int:isbn>', methods=['DELETE'])
def delete_book(isbn):
if(Book.delete_book(isbn)):
response = Response("",status=204)
return response
invalidBookObjectErrorMsg = {
"error":"Unable to Delete,as ISBN provided doesnot found in the List"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=404, mimetype='application/json') #Status404:Resource not found
return response
app.run(port=5000)
Open Terminal and run our server.
TERMINAL:
#TERMINAL:
Run server up
POSTMAN:
From POSTMAN fire a GET request to the books collection without a token, we will get an error that says we need a valid token to view this page.
#POSTMAN:
fire GET /books
#RESPONSE WINDOW:
{
"error": "Need a valid token"
}
#STATUS: 401 UNAUTHORIZED
Now do a GET request to our log in route(GET/at http://127.0.0.1:5000/login), and what this will do is it’ll return us a token that is valid for 100 seconds. If you pass that in to this GET request and do a Send, you can see the collection in response.
#POSTMAN:
fire GET/at http://127.0.0.1:5000/login
#RESPONSE WINDOW:
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE1NzQ5MzMxOTh9.uYmJYg5ufk2iKzv6Am5GA2G29hmo1fwaidRwVsvuuwg
#above token generated
fire GET/at http://127.0.0.1:5000/books?token=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE1NzQ5MzMxOTh9.uYmJYg5ufk2iKzv6Am5GA2G29hmo1fwaidRwVsvuuwg
#RESPONSE WINDOW:
{
"Books": [
{
"isbn": 753357,
"name": "NewBook1",
"price": 3.5
}
]
}
#STATUS 200 OK
You can check by modifying the token manually and send that, you will see the error: need a valid token to view this page, again.
6.3 Creating a Database for Users who are allowed API access
So far, we learned how to set up and create a JWT token. We were able to do things, such as add an expiration date to a token, as well as any other data we want to obtain between API calls. We also have the ability to receive tokens from requests, and determine whether the tokens to be received really are valid and authenticated. The goal now is to give tokens only to users who are authenticated. That way, only people we intend to use our API can access.
To do this create a new user table in our database that will store username and password. Whenever a client sends us a given username and password to our login route, we will check if the username and password combination exists in our database. If so, we will return to the client a token they can use. We will be using plain text passwords in our database and creating a basic user model to show you the inner details of how Flask integrates with JSON Web Tokens. On a production environment, we use an authentication library instead of coding everything from scratch. This is because libraries handle things, such as security and encoding passwords for us.
Lets code now, create a file called UserModel.py. Define a user model that can map a Python class that will create virtual users to an actual SQL table. And then create the initial database. This is very similar to what we did in the last module. Create db object, and this db object actually requires a couple of dependencies, and the very first one is the app object, which we import from our settings.py, and then the second thing is SQLAlchemy. From flask_sqlalchemy, import sqlalchemy. Define the model, which is basically a Python class that we’ll map to a SQL table. For this user class, we’ll inherit from the db.Model a declarative base class. And our User class will behave just like any other Python class, but it allows us to inherit a query method that could be used to run all types of searches in the database. This inherited query method allows us to access the SQL database and get all the data we need. Now, define the tablename as users and create columns.
Note: if we don’t define tablename, the class name that we use will be converted into a tablename. And there is a conversion process that happens, but we want to avoid that, which is why we are explicitly defining the tablename here.
Now specify the columns. The very first column we need is the ID, and this will act as our primary key set the primary_key property to be True and type Integer, use the same column method from the db object as we did before. The second column will hold the username define this as a string and it must be unique,so set unique equal to True. Make it non-nullable, by setting nullable=False. And the third column we have will hold the password, which is also a column that holds a string, same as the username make this password non-nullable. Now, let’s create our database.
Code Snippet_6.3-Creating a Database for Users who are allowed API access:
UserModel.py
#UserModel.py
from flask_sqlalchemy import SQLAlchemy
from settings import app
db = SQLAlchemy(app)
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password = db.Column(db.String(80), nullable=False)
TERMINAL:
Open the terminal and call Python. Now from our UserModel import everything and call the db.create_all method. Exit and do a cat of our database.db, we can see our user table and our books table, and we can see that user class we defined with a username, a password, the previous books class and the data that we currently have.
#TERMINAL:
$ ls
BookModel.py DataSanitizer_TestCase.py RSapp.py
database.db __pycache__ settings.py
$ touch UserModel.py #create a file
$ls
BookModel.py DataSanitizer_TestCase.py RSapp.py UserModel.py
database.db __pycache__ settings.py
$ subl UserModel.py #open file and paste the snippet given above
After coding the UserModel.py file, as instructed above.
#TERMINAL:
$ python
Python 3.6.8 (default)
[GCC 8.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from UserModel import *
>>> db.create_all()
>>> exit()
$ ls
BookModel.py DataSanitizer_TestCase.py RSapp.py UserModel.py
database.db __pycache__ settings.py
$ cat database.db
tableusers#CREATE TABLE users ( #<==userDB
id INTEGER NOT NULL, #<==newly created tables/columns
username VARCHAR(80) NOT NULL,
password VARCHAR(80) NOT NULL,
PRIMARY KEY (id),
UNIQUE (username)
))indexsqlite_autoindex_users_1users
#tablebooksbooks#CREATE TABLE books ( #<=books db
id INTEGER NOT NULL,
name VARCHAR(80) NOT NULL,
price FLOAT NOT NULL,
isbn INTEGER,
PRIMARY KEY (id)
NewBook1 #<=content
:
:
6.4 Creating Users ,who are allowed API access
Define a username_password_match function which takes a username and a password passed in, this method will be used as an API in our route to determine whether someone should get access or not, define this method to return a Boolean, if we find a match for the username and password combination that was sent to us, then we’ll get back a user object, otherwise if we don’t find a match we’ll get a non-user response. Define a getAllUsers method, this won’t be used publicly within the user API but helps us to test. Define new_user using our User constructor, add this to our database and commit this.
Runnable Code Snippet_6.4- Creating Users ,who are allowed API access:
# 6.4 Creating Users ,who are allowed API access
#UserModel.py
from flask_sqlalchemy import SQLAlchemy
from settings import app
db = SQLAlchemy(app)
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password = db.Column(db.String(80), nullable=False)
#printable representation of a given object
def __repr__(self):
return str({
'username': self.username,
'password': self.password
})
#username/password match ftn,
#acts as API to determine whether someone should get access
def username_password_match(_username, _password):
user = User.query.filter_by(username=_username).filter_by(password=_password).first()
if user is None:
return False
else:
return True
#getAllUsers method
def getAllUsers():
return User.query.all() #returns in the representation format given at top
#createUser method
def createUser(_username, _password):
new_user = User(username=_username, password=_password)
db.session.add(new_user)
db.session.commit()
TERMINAL:
Save the file, open a terminal and call Python. Now import everything from our UserModel. First get all the users to make sure that we get an empty list. Now check username_password_match by passing it different username and password, which should return false. Let’s create a user named user1, set its password to password. If we run the user.getAllUsers, we should see our new user that was created. Now check whether our username_password_match function actually works by passing it in user1 as username, and password as password this should return True, as expected.
#TERMINAL:
$ python
Python 3.6.8 (default)
[GCC 8.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from UserModel import *
>>> User.getAllUsers()
[] #<== empty
>>> User.username_password_match('a', 'b')
False
>>> User.createUser("user1", "password1")
>>> User.getAllUsers()
[{'username': 'user1', 'password': 'password1'}]
>>> User.username_password_match('user1', 'password1')
True
>>> User.username_password_match('user1', 'pass1word')
False
>>>
6.5 Having User Login to get a Token
Let’s use our UserModel API to validate that a given username and password passed in to our login route exists in our database. If it exists, we give client a JSON Web Token, otherwise return an unauthorized message back to the client. Edit the code as follows.
Code edit in app.py:
# 6.5 Having User Login to get a Token
#app.py
from UserModel import User #import User
#and edit the route directory as POST and add this snippet
@app.route('/login',methods=['POST'])
def get_token():
request_data = request.get_json()
username = str(request_data['username'])
password = str(request_data['password'])
match = User.username_password_match(username, password)
if match:
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
else:
return Response('', 401, mimetype='application/json')
Runnable Code Snippet_6.5 Having User Login to get a Token:
#app.py
from flask import Flask, jsonify,request,Response
from BookModel import *
from settings import *
import json #import Json
from settings import * #import all
import jwt, datetime #import jwt library which has JWT encode method
from UserModel import User #import User
app.config['SECRET_KEY'] = 'spinach'
@app.route('/login',methods=['POST'])
def get_token():
request_data = request.get_json()
username = str(request_data['username'])
password = str(request_data['password'])
match = User.username_password_match(username, password)
if match:
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
else:
return Response('', 401, mimetype='application/json')
#request body
#{
# 'name' : 'Newbook',
# 'price' : 456,
# 'isbn' : 369852
#}
#Sanitize the data received from Client
#data sanitizer
def validBookObject(bookObject):
#if entry already exists
for i in books:
if i["isbn"] == bookObject["isbn"]:
return False
else:
#if "keyword" in dictionary
if ("name" in bookObject and "price" in bookObject and "isbn" in bookObject):
return True
else:
return False
#Adding a POST route
#POST/books
@app.route('/books', methods=['POST'])
def add_book():
request_data = request.get_json()#get the req.data by get_json method
if(validBookObject(request_data)): #considering only validObject is sent
Book.add_book(request_data['name'], request_data['price'], request_data['isbn'])
response = Response("", status=201, mimetype='application/json')
response.headers['Location'] = "/books" + str(request_data['isbn'])
return response
else:
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
#GET/books?token=asok9f09rtwi978hs
@app.route('/books')
def get_all_books():
token = request.args.get('token')
try:
jwt.decode(token, app.config['SECRET_KEY'])
except:
return jsonify({'error': 'Need a valid token'}),401
return jsonify({'Books':Book.get_all_books()}) #from database
#GET/books/isbn
@app.route('/books/<int:isbn>')
def get_by_isbn(isbn):
return_value=Book.get_by_isbn(isbn)
return jsonify(return_value)
#PUT /books/5869
#{
# 'name': 'NewName',
# 'price': 456
#}
def valid_put_request_data(request_data):
if ("name" in request_data
and "price" in request_data):
return True
else:
return False
#PUT route
@app.route('/books/<int:isbn>', methods=['PUT'])
def replace_book(isbn):
request_data=request.get_json()
if(not valid_put_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
Book.replace_book(isbn, request_data['name'],request_data['price'])
response = Response("", status=204) #No content created
return response
# PATCH /books/isbn
#{
# 'name': 'UpdateNameAlone'
#}
# PATCH /books/isbn
#{
# 'price': 'UpdatePriceAlone'
#}
def valid_patch_request_data(request_data):
if ("name" in request_data
or "price" in request_data):
return True
else:
return False
@app.route('/books/<int:isbn>', methods=['PATCH'])
def update_book(isbn):
request_data = request.get_json() #we get the JSON data
if(not valid_patch_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error": "Invalid book object passed in request",
"helpstring": "Data should be in following format{'name': 'bookname', 'price':15}"
}
response = Response (json.dumps(invalidBookObjectErrorMsg),staus=404,mimetype='application/json')
return response
if("name" in request_data):
Book.update_book_name(isbn, request_data['name'])
if("price" in request_data):
Book.update_book_price(isbn, request_data['price'])
response = Response("",status=204) #204=Success
response.headers['Location'] = "/books/" + str(isbn)
return response
#DELETE /books/isbn
@app.route('/books/<int:isbn>', methods=['DELETE'])
def delete_book(isbn):
if(Book.delete_book(isbn)):
response = Response("",status=204)
return response
invalidBookObjectErrorMsg = {
"error":"Unable to Delete,as ISBN provided doesnot found in the List"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=404, mimetype='application/json') #Status404:Resource not found
return response
app.run(port=5000)
TERMINAL:
Fired requests from Postman and the expected outputs are given below.
#TERMINAL
$ python app.py
* Serving Flask app "settings"
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
127.0.0.1 - "GET /books HTTP/1.1" 401 - UNAUTHORIZED#<=tried entering /books route without any credentials
127.0.0.1 - "GET /login HTTP/1.1" 405 -METHOD NOT ALLOWED <=#<=tried entering /login route without any credentials
127.0.0.1 - "POST /login HTTP/1.1" 200 -OK #<==req body with a valid user {"username": user1, "password": password1) which is set in db already, and with contentType:json
#Got token generated
127.0.0.1-"GET/books?token=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE1NzQ5Mzc4NDB9.tvGlQ1yq1wfeLKUk-Ey1vO8U9NvosUC2LiqM0pUBfOw HTTP/1.1" 200 - OK #<=passed in token
127.0.0.1 -"POST /login HTTP/1.1" 401 - #<=gave invalid user in req body {"user": user2, "password": password1)
127.0.0.1-"GET/books?token=eyJ0eXAiOiJKV1QiLCJ1NiJ9.eyJleHAiOjE1NzQ5Mzc4NDB9.tvGlQ1yq1wfeLKUk-Ey1vO8U9NvosUC2LiqM0pUBfOw HTTP/1.1" 401 - #<==gave random token value
^C #stopped
6.6 Creating Decorators to Add API Authentication
Here we are going to migrate a lot of get_books code over and generalize it so that it can be used by any route. Very first thing we need to do is define our decorator, we define it as token_required which takes a function. Import wraps from functools, define our wrapper method, and this will take in the arguments and the keywords, and this wrapper method is going to replace the function that is being decorated. Get token from the request.args, try with this token to decode it if we get an exception return an error. Call @wraps. The code to be migrated and generalised is shown below.
Code Snippet -migrating and generalising:
#migrating and generalising so that it can be used by any route
#wrapper method replaces the function being decorated
from functools import wraps #for method token_required
def token_required(f):
@wraps(f)
def wrapper(*args, **Kwargs): #wrapper method takes arguments and keywords
token = request.args.get('token')
try:
jwt.decode(token, app.config['SECRET_KEY'])
except:
return jsonify({'error': 'Need a valid token'}),401
return wrapper
#GET/books?token=ak9f978hs
@app.route('/books')
@token_required
def get_all_books():
Runnable Code Snippet_6.6-Creating Decorators to Add API Authentication:
#app.py
from flask import Flask, jsonify,request,Response
from BookModel import *
from settings import *
import json #import Json
from settings import * #import all
import jwt, datetime #import jwt library which has JWT encode method
from UserModel import User #import User
from functools import wraps #for method token_required
app.config['SECRET_KEY'] = 'spinach'
@app.route('/login',methods=['POST'])
def get_token():
request_data = request.get_json()
username = str(request_data['username'])
password = str(request_data['password'])
match = User.username_password_match(username, password)
if match:
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
token = jwt.encode({'exp': expiration_date}, app.config['SECRET_KEY'],algorithm='HS256')
return token
else:
return Response('', 401, mimetype='application/json')
#request body
#{
# 'name' : 'Newbook',
# 'price' : 456,
# 'isbn' : 369852
#}
#Sanitize the data received from Client
#data sanitizer
def validBookObject(bookObject):
#if entry already exists
for i in books:
if i["isbn"] == bookObject["isbn"]:
return False
else:
#if "keyword" in dictionary
if ("name" in bookObject and "price" in bookObject and "isbn" in bookObject):
return True
else:
return False
#Adding a POST route
#POST/books
@app.route('/books', methods=['POST'])
def add_book():
request_data = request.get_json()#get the req.data by get_json method
if(validBookObject(request_data)): #considering only validObject is sent
Book.add_book(request_data['name'], request_data['price'], request_data['isbn'])
response = Response("", status=201, mimetype='application/json')
response.headers['Location'] = "/books" + str(request_data['isbn'])
return response
else:
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
#wrapper method replaces the function being decorated
def token_required(f):
@wraps(f)
def wrapper(*args, **kwargs): #wrapper method takes arguments and keywords
token = request.args.get('token')
try:
jwt.decode(token, app.config['SECRET_KEY'])
return f(*args, **kwargs)
except:
return jsonify({'error': 'Need a valid token'}),401
return wrapper
#GET/books?token=asok9f09rtwi978hs
@app.route('/books')
@token_required
def get_all_books():
return jsonify({'Books':Book.get_all_books()}) #from database
#GET/books/isbn
@app.route('/books/<int:isbn>')
def get_by_isbn(isbn):
return_value=Book.get_by_isbn(isbn)
return jsonify(return_value)
#PUT /books/5869
#{
# 'name': 'NewName',
# 'price': 456
#}
def valid_put_request_data(request_data):
if ("name" in request_data
and "price" in request_data):
return True
else:
return False
#PUT route
@app.route('/books/<int:isbn>', methods=['PUT'])
def replace_book(isbn):
request_data=request.get_json()
if(not valid_put_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error" : "Invalid book object in request",
"helpString" : "Pls pass Data in similar format {'name':'bookname', 'price':5.9, 'isbn':852569}"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=400,mimetype='application/json')
return response
Book.replace_book(isbn, request_data['name'],request_data['price'])
response = Response("", status=204) #No content created
return response
# PATCH /books/isbn
#{
# 'name': 'UpdateNameAlone'
#}
# PATCH /books/isbn
#{
# 'price': 'UpdatePriceAlone'
#}
def valid_patch_request_data(request_data):
if ("name" in request_data
or "price" in request_data):
return True
else:
return False
@app.route('/books/<int:isbn>', methods=['PATCH'])
def update_book(isbn):
request_data = request.get_json() #we get the JSON data
if(not valid_patch_request_data(request_data)):
invalidBookObjectErrorMsg = {
"error": "Invalid book object passed in request",
"helpstring": "Data should be in following format{'name': 'bookname', 'price':15}"
}
response = Response (json.dumps(invalidBookObjectErrorMsg),staus=404,mimetype='application/json')
return response
if("name" in request_data):
Book.update_book_name(isbn, request_data['name'])
if("price" in request_data):
Book.update_book_price(isbn, request_data['price'])
response = Response("",status=204) #204=Success
response.headers['Location'] = "/books/" + str(isbn)
return response
#DELETE /books/isbn
@app.route('/books/<int:isbn>', methods=['DELETE'])
def delete_book(isbn):
if(Book.delete_book(isbn)):
response = Response("",status=204)
return response
invalidBookObjectErrorMsg = {
"error":"Unable to Delete,as ISBN provided doesnot found in the List"
}
response = Response(json.dumps(invalidBookObjectErrorMsg),status=404, mimetype='application/json') #Status404:Resource not found
return response
app.run(port=5000)
TERMINAL:
Open up Terminal and Postman. Go to our books route and check that it ask for a token and ensure. Now go into our login route and pass in our username and password, now we should get a token back. Send this token over to get the book collection back.
#TERMINAL:
$ python app.py
* Serving Flask app "settings"
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
127.0.0.1 - "GET /books HTTP/1.1" 401 -#<=without credentials
127.0.0.1 - "POST /login HTTP/1.1" 200 -#<=with credentials #we get token generated
127.0.0.1-"GET/books?token=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE1NzQ5NDEyNzN9.KPLNPo98JAlpNwwwvU7_qjXnNSWnlFeX13x-EUqkGsU HTTP/1.1" 200 - #<=entered with token
127.0.0.1-"GET/books?token=eyJ0eXAiOiJKV1QiLCJhHAiOjE1NzQ5NDEyNzN9.KPLNPo98JAlpNwwwvU7_qjXnNSWnlFeX13x-EUqkGsU HTTP/1.1" 401 - #<= tried with random token
127.0.0.1 -"POST /login HTTP/1.1" 401 -#<=tried with random credential
^C #stopped
6.7 Adding Authentication for all our routes using Decorators
Now we add wrapper infront of all the methods and check authentication. We defined a decorator that allows us to acquire an access token for a client if they want to use any part of our API, but we made it so that GET requests of a resource require an access token, but typically for rest APIs we allow read access to resources, but restrict write access. We edit this now and make it so any POST, PUT, or PATCH request requires a token, while any GET request does not. So remove the token_required method from the get_books route and put token_required method in PUT, PATCH and DELETE routes.
Hope you can test this on your own, if you have followed all the modules.Please try it and let me know.
With this we have come to the end of this writeup, so far we added authentication to our API. We learnt how to create and validate tokens to only allow clients who have a valid token to be able to gain access to our resource. We then made it so clients actually needed to register in order to use our API. When a client tries to log in, their credentials would be checked against the database, and only if they have an account they will be able to get a token that will allow them to use our API. We then learnt how to use decorators to abstract all this functionality into one line of code.
To sum up this course., we started from the basics and created our first Hello World Flask application. From there, we then built out routes for GET, POST, PUT, PATCH, and DELETE requests. Once all that was done, we then migrated all of our data to a database and used Flask’s SQLAlchemy as our ORM. We covered the basics and foundations of JWT. Ultimately we had fun writing some Flask code : )
Thank you all and visit my other writeups and suggest me for any improvements.
love,
<R.S>
For Complete Sourcecode visit my Github repo
RS-codes
RS-codes 
One thought on “First time with Flask-Python -Module6”