Flask: User Authentication

Ege
Ege
Sep 8, 2018 · 7 min read

Hi,

Today I will write about how to build user authentication system with Flask microframework.

Before I dive into code, I would like to show the architecture of directories:

.
|__ _database
| |-- db.py
|__ _models
| |-- user.py
|__ _resources
| |-- user.py
|__ _app.py

First of all, I will start with database/db.py file. In this file, we will initialize database object. For database, we will use Flask-SQLAlchemy which is an extension of Flask. It uses SQLite Database. By using this extension, we are able to use ORM(Object Relational Mapping), so that we do not have to write SQL statements like “SELECT * FROM users” or “INSERT INTO users …”


Now, it is time to talk about models/user.py file. Name of our model will be UserModel. This model will represent a user. Each user will have a unique id, username and a password. In other words, our table in database will have 3 columns: id, username, password.

__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80))
password = db.Column(db.String())

SQLAlchemy will force us to define a table name. But in Flask-SQLAlchemy it is optional, it gives a default table name. But if you want to change table name, you should override __tablename__. As I said, we have 3 columns, to define a column, you should use db.Column. First argument of the db.Column will be the type of the column. For id column, we also specify that it is Primary Key. We do not have to make this id as auto-increment, Flask-SQLAlchemy will automatically set it to auto-increment if this Primary Key is not a Foreign Key.

def __init__(self, username, password):
self.username = username
self.password = password

Our UserModel class needs a constructor. We pass username and password as arguments. We do not have to pass an id since it will be auto incremented.

def json(self):
return {
"id": self.id,
"username": self.username
}, 200

This method will return our UserModel Object in JSON format.

# Method to save user to DB
def save_to_db(self):
db.session.add(self)
db.session.commit()
# Method to remove user from DB
def remove_from_db(self):
db.session.delete(self)
db.session.commit()

These two methods are for database manipulation. Inserting data into database has 3 steps:

  1. Create UserModel object.
  2. Add it to session.
  3. Commit the session.

Removing data from database has similar 3 steps, instead of adding, we will delete it from session.

# Class method which finds user from DB by username
@classmethod
def find_user_by_username(cls, username):
return cls.query.filter_by(username=username).first()
# Class method which finds user from DB by id
@classmethod
def find_user_by_id(cls, _id):
return cls.query.filter_by(id=_id).first()

These are the helper method of UserModel class. To find a user by its username, we have to query the database. Luckily we do not have to write “SELECT * FROM users WHERE username == …”. From our model, we will use query attribute and filter the results by giving a value for username column. Then, we will get the first result. This first result is a UserModel object. To find user by its id, we will do the same thing.

Full code of this model. We pass db.Model to UserModel. If we did not do this, then Flask-SQLAlchemy won’t be able to recognize this class as a model.


Now, it is time to talk about resources/user.py file.In this file, we use Flask-RESTful extension to build REST API very easily. In addition to that, we will also use Flask-JWT-Extended extension to use JSON Web Tokens. JWT is important to protect our API from foreigners.

_user_parser = reqparse.RequestParser()
_user_parser.add_argument(
"username",
type=str,
required=True,
help="This field cannot be blank"
)
_user_parser.add_argument(
"password",
type=str,
required=True,
help="This field cannot be blank"
)

First of all, we define a Request Parser. In our request to register and login, we send 2 information to our API: username and password. With reqparse, we parse this request like a Python Dictionary.

class User(Resource):
def get(self, user_id):
user = UserModel.find_user_by_id(user_id)
if user:
return user.json()
return {
"message": "User not found!"
}, 404
@fresh_jwt_required
def delete(self, user_id):
user = UserModel.find_user_by_id(user_id)
if user:
user.remove_from_db()
return {
"message": "User deleted!"
}, 200
return {
"message": "User not found!"
}, 404

We have defined a class: User. This class is a Resource. So if someone sends a request to our API, it will be interact with this Resource, not the Model we have defined earlier. There are some HTTP Requests as GET, POST, DELETE, PUT. In a resource, we can define what is going to be happen in each request. For example in this User Resource, we have defined GET and DELETE methods.

  • When GET request received, we check the database with the given user id. As you remember, we wrote some helper methods in our UserModel Class. Corresponding method will return UserModel object if user exists, if not it will return None. If user exists, we will return this object by converting it to JSON format by calling the json() method we wrote.
  • When DELETE request received, first we check the database with the given user id. If user exists, we will delete it by the helper method which we wrote in UserModel.

I will talk about the fresh_jwt_required annotation later on.

class UserRegister(Resource):
def post(self):
data = _user_parser.parse_args()
if UserModel.find_user_by_username(data["username"]):
return {
"message": "User exists!"
}, 400
user = UserModel(data["username"], hashlib.sha256(data["password"].encode("utf-8")).hexdigest())
user.save_to_db()
return {
"message": "User {} created!".format(data["username"])
}

Our second Resource is for registering new user. It has POST method. So client will call POST method with username and password arguments. We parse these arguments by using our parser. First we check our database if there is a user with the given username. If yes, we return a message which says that a user exists with that username. If not, we create UserModel object with the given username and password. Before putting password into the database, we hash it for security. You can do this process on client, before sending this request. Then we use the helper method in UserModel in order to save this user to database.

class UserLogin(Resource):
def post(self):
data = _user_parser.parse_args()
user = UserModel.find_user_by_username(data["username"])if user and user.password == hashlib.sha256(data["password"].encode("utf-8")).hexdigest():
access_token = create_access_token(identity=user.id, fresh=True) # Puts User ID as Identity in JWT
refresh_token = create_refresh_token(identity=user.id) # Puts User ID as Identity in JWT
return {
"access_token": access_token,
"refresh_token": refresh_token
}, 200
return {
"message": "Invalid credentials!"
}, 401

Third Resource is for logging in. It also contains POST method. It will parse the username and password. It will check the user in database. If exists, then it will check if passwords match. If yes, we will create 2 different tokens(methods to create these tokens are built in):

  • Access Token: This JWT used to identify user. We return this token to the user after login. With this token, our API can understand that this client has logged in. We put extra information as identity, giving the users id.
  • Refresh Token: This special kind of token is used to create a new fresh access token.
class TokenRefresh(Resource):
@jwt_refresh_token_required
def post(self):
current_user_id = get_jwt_identity() # Gets Identity from JWT
new_token = create_access_token(identity=current_user_id, fresh=False)
return {
"access_token": new_token
}, 200

And the last Resource is to create a new access token, has POST method. As you remember, after user logs in, we were creating access token by putting an extra information as identity. And identity was keeping user id. By calling get_jwt_identity() we obtain that users id. After that, we create a new token, but this time, we mark fresh option as False.

DELETE method User Resource had a annotation: fresh_jwt_required. While sending this request, we should also send access token which is fresh. If not, we will have an error. We could also ask for a token which is not fresh. But deleting a user is a critical operation, so we should do this with a fresh token, which can be obtained right after login. But for example, you have an API for a bookstore and GET method will return information about a book, which is not a critical operation. You have logged in and got a fresh JWT. But this token expires after 5 minutes. While user checking different books, it would be odd to login in every 5 minutes. To inhibit this situation, GET method also requires a JWT but this time, this token does not have to be fresh. So when access token expires, we will use TokenRefresh resource to obtain a new access token, which can be used to GET information about a book.

Full code for Resources which are related to User operations.


Finally, I will share app.py file which mostly contains configurations.

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get
("DATABASE_URL", "sqlite:///data.db")
app.config[
"SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["PROPAGATE_EXCEPTIONS"] = True
app.secret_key = "v3ry_s3cr3t_k3y"

We initialize app Object as a Flask Object. These are the configurations for the Flask app.

api = Api(app)jwt = JWTManager(app)

api object is an Api object which supported by Flask-RESTful. JWTManager has supported by Flask-JWT-Extended. We have defined that this Flask app has an api and jwt.

@jwt.expired_token_loader
def expired_token_callback():
return jsonify(
{
"description": "Token has expired!",
"error": "token_expired"
}, 401
)
@jwt.invalid_token_loader
def invalid_token_callback():
return jsonify(
{
"description": "Signature verification failed!",
"error": "invalid_token"
}, 401
)
@jwt.unauthorized_loader
def unauthorized_loader_callback(error):
return jsonify(
{
"description": "Access token not found!",
"error": "unauthorized_loader"
}, 401
)
@jwt.needs_fresh_token_loader
def fresh_token_loader_callback():
return jsonify(
{
"description": "Token is not fresh. Fresh token needed!",
"error": "needs_fresh_token"
}, 401
)

These are the callbacks which Flask-JWT-Extended supports. I wont explain them, you can understand these callbacks by reading their description.

api.add_resource(User, "/user/<int:user_id>")
api.add_resource(UserRegister, "/register")
api.add_resource(UserLogin, "/login")

This part is very important because routing happens in this part. We wrote our resources, but we haven’t defined the end-points of these resources. In this part, we did it. To call methods in User, we should call “/user/<int:user_id”> end-point. Example of this end-point can be “/user/1”. If we call GET method with this end-point, it will return information about the user which has id 1.

if __name__ == '__main__':
from database.db import db
db.init_app(app) @app.before_first_request
def create_tables():
db.create_all()
app.run(port=5000, debug=True)

In this part, we define the database of this app, which was in db.py file. To create tables in out database, we call create_all method of db object. Last but not least, we run our app, which will listen port 5000. By setting debug = True, we make error message better to understand.


This is the end of this story. I hope I have helped you :)

Ege

Written by

Ege

CS Student @SabanciUni

Welcome to a place where words matter. On Medium, smart voices and original ideas take center stage - with no ads in sight. Watch
Follow all the topics you care about, and we’ll deliver the best stories for you to your homepage and inbox. Explore
Get unlimited access to the best stories on Medium — and support writers while you’re at it. Just $5/month. Upgrade