In this lab we shall return to building application server routes. In order to test them we will need to use postman
Just as in lab 1 follow the link below to open repl.it. Fork the project then sign in.
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.
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.
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.
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:
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.
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.
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.
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.
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 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. |
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.
Create a /todo POST route which does the following:
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.
Create a /todo GET route which does the following:
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.
Create a /todo/<id> GET route which does the following:
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.
Create a /todo/<id> PUT route which does the following:
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
Create a /todo/<id> DELETE route which does the following:
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.