In this lab we shall return to building application server routes. In order to test them we will need to use postman

Task 1.1

Just as in lab 1 follow the link below to open repl.it. Fork the project then sign in.

Open Lab 3 Workspace

Task 1.2

Open this lab's API documentation and import the postman collection by first Opening Postman on the lab machine leaving it on the login page.


Next click the following link.

Open Todo API Postman Doc

Then click on the "Run in Postman" Button

Select Postman for Windows


Then click in on "Open in Postman"


Postman should then open and import the Lab 5 Collection.

Next click on the "Environment Quick Look" button to edit postman variables.

Next click on "Add"

We use this screen to set variables which are used in different parts of postman. This allows us to run postman on your repl workspace by setting a "host" variable to the url of your Running Repl Web Application.

Rename the environment to "Lab 5 Repl" then create a variable
host and set the values to your repl's app url.



Then press save and select "Lab 5 Repl" in the environments dropdown.


We want to expand our todo application to cater for multiple user groups. Sometimes applications have different user classes that are authorized to perform different actions. This lab continues from Lab 2 except the following user classes have been made.

Regular User: Previously the user model, can create edit delete their own todos

Admin User: Can access all todos in the application.

SQLAlchemy has different ways of performing inheritance based on our needs. More info here.

The workspace has been updated with the relevant changes for RegularUser.

Note the addition of the various methods such as:

Also note how the init command has been updated to insert data in the database by parsing todos.json

#load todo data from csv file
with open('todos.csv') as file:
   reader = csv.DictReader(file)
   for row in reader:
     new_todo = Todo(text=row['text']) #create object
     #update fields based on records
     new_todo.done = True if row['done'] == 'true' else False
     new_todo.user_id = int(row['user_id'])
     db.session.add(new_todo) #queue changes for saving
   db.session.commit() 
   #save all changes OUTSIDE the loop

Notice the use of the int() function to parse our string data.

Task 2.1

Update models.py and update the Admin class as follows

class Admin(User):
  __tablename__ = 'admin'
  staff_id = db.Column(db.String(120), unique=True)
  __mapper_args__ = {
      'polymorphic_identity': 'admin',
  }

  def get_all_todos_json(self):
    todos = Todo.query.all()
    if todos:
      return [todo.get_json() for todo in todos]
    else:
      return []

  def __init__(self, staff_id, username, email, password):
    super().__init__(username, email, password)
    self.staff_id = staff_id

  def get_json(self):
    return {
        "id": self.id,
        "username": self.username,
        "email": self.email,
        "staff_id": self.staff_id,
        "type": self.type
    }

  def __repr__(self):
    return f'<Admin {self.id} : {self.username} - {self.email}>'

The new property __mapper_args lets us set the value of the type property defined on the parent of the model.

Now we have a new model we need to reinitialize our db. Before doing so lets change our init command to add the admin Pam.

Task 2.2

Initialize the db

Now run the command "list-todos" to list the todos in the application

In many applications it is typical to restrict access to features from unauthorized parties. If your application deals with user data then users should be able to only manipulate the data that belong to them. This is achieved by applying the following concepts

In this lab we shall demonstrate token based authentication where users must login to receive a token that is used in every subsequent request to restricted resources.

We will be using Flask JWT Extended which implements the JSON Web Token scheme. Flask JWT Extended is relatively easy to set up we just need to to the following:

Task 3.1

Create the following login function.

def login_user(username, password):
  user = User.query.filter_by(username=username).first()
  if user and user.check_password(password):
    token = create_access_token(identity=username)
    response = jsonify(access_token=token)
    set_access_cookies(response, token)
    return response
  return jsonify(message="Invalid username or password"), 401

If the username given exists and the user password matches then a token is generated based on the username else None is returned.

Flask JWT Extended provides a useful create_access_token() function that we shall use to create JSON Web Tokens based on a given username in app.y.

We use set_access_cookies() to create a cookie in the response to be saved in the browser.

Task 3.2

Add the following to app.py

@app.route('/login', methods=['POST'])
def user_login_view():
  data = request.json
  response = login_user(data['username'], data['password'])
  if not response:
    return jsonify(message='bad username or password given'), 403
  return response

We use request.json to retrieve the json data sent in the request by postman then pass their username and password fields to the user_login().

By default, routes in flask work with the GET request method. In order to receive encrypted form data we need to use the POST request method hence POST is specified in the methods parameter.

Finally you can execute the login route by opening the login request from the collection in the sidebar and clicking the Send button. Noice the method is set to POST.

You would notice the api responded with an access_token which means that the login was successful. You can inspect the JSON body of the request by clicking on the body tab.

If you click the body dropdown of the response to cookies you can then see a cookie was also created that stores the token.

Next we shall add a route which requires authenticated access.

Task 3.3

Add the following route to app.py

@app.route('/identify')
@jwt_required()
def identify_view():
  username = get_jwt_identity()
  user = User.query.filter_by(username=username).first()
  if user:
    return jsonify(user.get_json())
  return jsonify(message='Invalid user'), 403

get_jwt_identity was imported from Flask-JWT and it returns the decoded username object from the token given in the request.

The @jwt_required() decorator restricts access to this route to authenticated users if a jwt token is not present in the request (cookie) it will return a 401 error.

Open the "Identify User" request in Postman but before running, select Authorization in the dropdown

Now run the Identify user request in postman you should see the following if you ran login earlier.

If you switch the request tab to headers you can see that additional information was sent in the request due to the cookie stored earlier.

Specifically the information is the JWT access_token created at the login step. Hence a request can be made to this @jwt_required route and the user has been correctly identified from the token.


From the previously mentioned cookies view; you can delete the cookie and try running the request again. You will notice that the user no longer has access to the route.

Task 3.4

We can't expect users to manually delete cookies so we should make the following logout route in app.py to do it for them.

@app.route('/logout', methods=['GET'])
def logout():
  response = jsonify(message='Logged out')
  unset_jwt_cookies(response)
  return response

We just added routes to login our pre-existing users but applications need a way to create users by registering/signing up.

Task 4

Update app.py to add the following signup route..

@app.route('/signup', methods=['POST'])
def signup_user_view():
  data = request.json
  try:
    new_user = RegularUser(data['username'], data['email'], data['password'])
    db.session.add(new_user)
    db.session.commit()
    return jsonify(message=f'User {new_user.id} - {new_user.username} created!'), 201
  except IntegrityError:
    db.session.rollback()
    return jsonify(message='Username already exists'), 400

Now try executing the signup route in postman.

The new user should be created. Note because we used Single Table Inheritance we would not have admins and users with the same username.

RESTful APIs

RESTful is an architectural style for building out the urls of an application server. It specifies which HTTP methods should be used according to a desired CRUD operation.

In this lab we shall implement the following API Specification

Route Name

(Authorized

User Class)

HTTP method

Description

Sign Up

POST

Creates a user and logs them in if the user does not exist.
Returns an error message if user exists

Login

POST

Logs in user, returns a token if credentials are correct

Get Todos

(Regular User)

GET

Returns all of a user's todos

Get Todo

(Regular User)

GET

Retrieves a todo if the user is authorized to access it

Create Toto

(Regular User)

POST

Creates a Todo if the user is authorized

Update Todo

(Regular User)

PUT

Updates a Todo if the user is authorized to access it

Delete Todo

(Regular User)

DELETE

Deletes a Todo if the user is authorized to access it

Next we want to make a route which allows logged in users to create a Todo.

Task 5.1

Create a /todo POST route which does the following:

  1. Restrict it to logged in regular users
  2. Store the post data in a variable called data
  3. Create a todo object using add_todo()
  4. Return the id of the todo object and a status code of 201

Solution

@app.route('/todos', methods=['POST'])
@login_required(RegularUser)
def create_todo_view():
  data = request.json
  username = get_jwt_identity()
  user = RegularUser.query.filter_by(username=username).first()
  new_todo = user.add_todo(data['text'])
  return jsonify(message=f'todo {new_todo.id} created!'), 201


Because we want only regular users to access this route we need a more specific decorator than @jwt_required. Only JWT's of regular users are allowed so we use our custom made @login_required() decorator and we pass in the desired user class RegularUser.

This ensures that only Regular Users can make this request.


If you execute the "Create Todo" request the response would be the id of the newly created Todo.


Task 5.2

Create a /todo GET route which does the following:

  1. Restrict access to logged in users
  2. Retrieve the user's todos
  3. Convert the todo objects to dictionaries
  4. Return the user's todos as json

Solution

@app.route('/todos', methods=['GET'])
@jwt_required()
def get_todos_view():
  # get the user object of the authenticated user
  user = RegularUser.query.filter_by(username=get_jwt_identity()).first()
  # converts todo objects to list of todo dictionaries
  todo_json = [ todo.get_json() for todo in user.todos ]
  return jsonify(todo_json), 200

Run the "Get Todos" Request

Only the todos of the logged in user should be shown.

Task 5.3

Create a /todo/<id> GET route which does the following:

  1. Restrict access to logged in users
  2. Get the todo by the id specified in the route parameter
  3. Check if the todo belongs to the logged in user
  4. Returns an error message if not found or unauthorized
  5. Returns a json representation of the todo if found

Solution

@app.route('/todos/<int:id>', methods=['GET'])
@jwt_required()
def get_todo_view(id):
  todo = Todo.query.get(id)

  # must check if todo belongs to the authenticated user
  if not todo or todo.user.username != get_jwt_identity():
    return jsonify(error="Bad ID or unauthorized"), 401
  
  return jsonify(todo.get_json()), 200

Run the ‘GET Todo' request. Notice how the route parameter of 5 for the id is set in the params tab.

Task 5.4

Create a /todo/<id> PUT route which does the following:

  1. Restrict access to logged in users
  2. Retrieves the json data from the request
  3. Get the todo by the id specified in the route parameter
  4. Check if the todo belongs to the logged in user
  5. Returns an error message if not found or unauthorized
  6. If found updates to object using the request data
  7. Saves the object to in the database
  8. Return updated in the response

Solution

@app.route('/todos/<int:id>', methods=['PUT'])
@login_required(RegularUser)
def edit_todo_view(id):
  data = request.json
  user = RegularUser.query.filter_by(username=get_jwt_identity()).first()

  todo = Todo.query.get(id)

  # must check if todo belongs to the authenticated user
  if not todo or todo.user.username != get_jwt_identity():
    return jsonify(error="Bad ID or unauthorized"), 401

  user.update_todo(id, data['text'])
  return jsonify(message=f"todo updated to '{data['text']}'!"), 200

Run the "Update Todo" request

Task 5.5

Create a /todo/<id> DELETE route which does the following:

  1. Restrict access to logged in users
  2. Get the todo by the id specified in the route parameter
  3. Check if the todo belongs to the logged in user
  4. Returns an error message if not found or unauthorized
  5. Deletes the object from the database
  6. Return updated in the response

Solution

@app.route('/todos/<int:id>', methods=['DELETE'])
@jwt_required()
def delete_todo_view(id):
  user = RegularUser.query.filter_by(username=get_jwt_identity()).first()
  todo = Todo.query.get(id)

  # must check if todo belongs to the authenticated user
  if not todo or todo.user.username != get_jwt_identity():
    return jsonify(error="Bad ID or unauthorized"), 401

  user.delete_todo(id)
  return jsonify(message="todo deleted!"), 200

Run the "Delete Todo" request with an appropriate id.

You should no longer be able to retrieve that todo

The API Specification has finally been implemented!!

Congratulations for making it to the end. You have just implemented a RESTful API with authentication via JSON Web Tokens. This lab forms the core of the server side programming done in the course and is VERY important with respect to your assessments.

Lab 3 Completed

References & Additional Reading