DEV Community

Cover image for Building a User Authentication API using Python Flask and MySQL
Andrew Eze
Andrew Eze

Posted on • Edited on

Building a User Authentication API using Python Flask and MySQL

#python #flask #api #mysql

Recently, I decided to learn Python, as part of learning I built a remote jobs (remote4africa) platform using Python Flask. In this post, I will show you step by step process for building a user authentication API using Python (Flask) and MySQL. The application will allow users to register and verify their email through an OTP Code sent to their email.

Note: This post assumes you have an understanding of Python Flask and MySQL.

Flask is a micro web framework written in Python. It allows python developers to build APIs or full web app (frontend and Backend).

When building a user authentication system, you should consider security and ease of use.
On security, you should not allow users to use weak passwords, especially if you are working on a critical application. You should also encrypt the password before storing in your database.

To build this I used the following dependencies:
cerberus: For validation
alembic: For Database Migration and Seeding
mysqlclient: For connecting to MySQL
Flask-SQLAlchemy, flask-marshmallow and marshmallow-sqlalchemy: For database object relational mapping
pyjwt: For JWT token generation
Flask-Mail: For email sending
celery and redis: For queue management

So let's get started

Step 1: Install and Set up your Flask Project.

You can follow the guide at Flask Official Documentation site or follow the steps below. Please ensure you have python3 and pip installed in your machine.

$ mkdir flaskauth $ cd flaskauth $ python3 -m venv venv $ . venv/bin/activate $ pip install Flask # pip install requirements.txt
Enter fullscreen mode Exit fullscreen mode

You can install your all the required packages together with Flask at once using a requirements.txt file. (This is provided in the source code).

The Python Flask App folder Structure
It is always advisable to use the package pattern to organise your project.

/yourapplication /yourapplication __init__.py /static style.css /templates layout.html index.html login.html ...
Enter fullscreen mode Exit fullscreen mode

Add a pyproject.toml or setup.py file next to the inner yourapplication folder with the following contents:

pyproject.toml

[build-system] requires = ["setuptools"] build-backend = "setuptools.build_meta" [project] name = "yourapplication" description = "yourapplication description" readme = "README.rst" version="1.0.0" requires-python = ">=3.11" dependencies = [ "Flask", "SQLAlchemy", "Flask-SQLAlchemy", "wheel", "pyjwt", "datetime", "uuid", "pytest", "coverage", "python-dotenv", "alembic", "mysqlclient", "flask-marshmallow", "marshmallow-sqlalchemy", "cerberus", "Flask-Mail", "celery", "redis" ]
Enter fullscreen mode Exit fullscreen mode

setup.py

from setuptools import setup setup( name='yourapplication', packages=['yourapplication'], include_package_data=True, install_requires=[ 'flask', ], py_modules=['config'] )
Enter fullscreen mode Exit fullscreen mode

You can then install your application so it is importable:

$ pip install -e .
Enter fullscreen mode Exit fullscreen mode

You can use the flask command and run your application with the --app option that tells Flask where to find the application instance:

$ flask –app yourapplication run
Enter fullscreen mode Exit fullscreen mode

In our own case the application folder looks like the following:

/flaskauth /flaskauth __init__.py /models /auth /controllers /service /queue /templates config.py setup.py pyproject.toml /tests .env ...
Enter fullscreen mode Exit fullscreen mode

Step 2: Set up Database and Models

Since this is a simple application, we just need few tables for our database:

  • users
  • countries
  • refresh_tokens

We create our baseModel models/base_model.py

from flask_sqlalchemy import SQLAlchemy from flask_marshmallow import Marshmallow from sqlalchemy.ext.declarative import declared_attr from flaskauth import app _PLURALS = {"y": "ies"} db = SQLAlchemy() ma = Marshmallow(app) class BaseModel(object): @declared_attr def __tablename__(cls): name = cls.__name__ if _PLURALS.get(name[-1].lower(), False): name = name[:-1] + _PLURALS[name[-1].lower()] else: name = name + "s" return name
Enter fullscreen mode Exit fullscreen mode

For Users table, we need email, first_name, last_name, password and other fields shown below. So we create the user and refresh token model models/user.py

from datetime import datetime from flaskauth.models.base_model import BaseModel, db, ma from flaskauth.models.country import Country class User(db.Model, BaseModel): __tablename__ = "users" id = db.Column(db.BigInteger, primary_key=True) email = db.Column(db.String(120), unique=True, nullable=False) password = db.Column(db.String(200), nullable=True) first_name = db.Column(db.String(200), nullable=False) last_name = db.Column(db.String(200), nullable=False) avatar = db.Column(db.String(250), nullable=True) country_id = db.Column(db.Integer, db.ForeignKey('countries.id', onupdate='CASCADE', ondelete='SET NULL'), nullable=True) is_verified = db.Column(db.Boolean, default=False, nullable=False) verification_code = db.Column(db.String(200), nullable=True) created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.utcnow) deleted_at = db.Column(db.DateTime, nullable=True) country = db.relationship('Country', backref=db.backref('users', lazy=True)) refresh_tokens = db.relationship('RefreshToken', backref=db.backref('users', lazy=True)) class RefreshToken(db.Model, BaseModel): __tablename__ = "refresh_tokens" id = db.Column(db.BigInteger, primary_key=True) token = db.Column(db.String(200), unique=True, nullable=False) user_id = db.Column(db.BigInteger, db.ForeignKey(User.id, onupdate='CASCADE', ondelete='CASCADE'), nullable=False) expired_at = db.Column(db.DateTime, nullable=False) created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.utcnow)
Enter fullscreen mode Exit fullscreen mode

We also create the country model

from flaskauth.models.base_model import BaseModel, db from flaskauth.models.region import Region class Country(db.Model, BaseModel): __tablename__ = "countries" id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(100), unique=True, nullable=False) code = db.Column(db.String(3), unique=True, nullable=False)
Enter fullscreen mode Exit fullscreen mode

For migration and seeding (creating the tables on the database and importing default data), we will use alembic. I will show you how to do this later.

Step 3: Create the __init__.py file

In this file we will initiate the flask application, establish database connection and also set up the queue. The init.py file makes Python treat directories containing it as modules.

import os from flask import Flask, make_response, jsonify from jsonschema import ValidationError def create_app(test_config=None): app = Flask(__name__, instance_relative_config=True) if test_config is None: app.config.from_object('config.ProductionConfig') else: app.config.from_object('config.DevelopmentConfig') # ensure the instance folder exists try: os.makedirs(app.instance_path) except OSError: pass return app app = create_app(test_config=None) from flaskauth.models.base_model import db, BaseModel db.init_app(app) from flaskauth.models.user import User from celery import Celery def make_celery(app): celery = Celery(app.name) celery.conf.update(app.config["CELERY_CONFIG"]) class ContextTask(celery.Task): def __call__(self, *args, **kwargs): with app.app_context(): return self.run(*args, **kwargs) celery.Task = ContextTask return celery celery = make_celery(app) from flaskauth.auth import auth from flaskauth.queue import queue from flaskauth.controllers import user app.register_blueprint(auth, url_prefix='/auth') app.register_blueprint(queue) @app.route("/hello") def hello_message() -> str: return jsonify({"message": "Hello It Works"}) @app.errorhandler(400) def bad_request(error): if isinstance(error.description, ValidationError): original_error = error.description return make_response(jsonify({'error': original_error.message}), 400) # handle other "Bad Request"-errors # return error return make_response(jsonify({'error': error.description}), 400)
Enter fullscreen mode Exit fullscreen mode

This file loads our application by calling all the models and controllers needed.

The first function create_app is for creating a global Flask instance, it is the assigned to app

app = create_app(test_config=None)
Enter fullscreen mode Exit fullscreen mode

We then import our database and base model from models/base_model.py

We are using SQLAlchemy as ORM for our database, we also use Marshmallow for serialization/deserialization of our database objects.
With the imported db, we initiate the db connection.

Next we use the make_celery(app) to initiate the celery instance to handle queue and email sending.

Next we import the main parts of our applications (models, controllers and other functions).

app.register_blueprint(auth, url_prefix='/auth') app.register_blueprint(queue)
Enter fullscreen mode Exit fullscreen mode

The above will register the queue and auth blueprints. In the auth blueprint we handle all routes that starts with /auth

@app.route("/hello") def hello_message() -> str: return jsonify({"message": "Hello It Works"})
Enter fullscreen mode Exit fullscreen mode

We will use this to test that our application is running.

The bad_request(error): function will handle any errors not handled by our application.

The Authentication

To handle authentication we will create a file auth/controllers.py
This file has the register function that will handle POST request. We also need to handle data validation to ensure the user is sending the right information.

from werkzeug.security import generate_password_hash, check_password_hash from flaskauth import app from flaskauth.auth import auth from flask import request, make_response, jsonify, g, url_for from datetime import timedelta, datetime as dt from flaskauth.models.user import db, User, RefreshToken, UserSchema from sqlalchemy.exc import SQLAlchemyError from cerberus import Validator, errors from flaskauth.service.errorhandler import CustomErrorHandler from flaskauth.queue.email import send_email from flaskauth.service.tokenservice import otp, secret, jwtEncode from flaskauth.service.api_response import success, error @auth.route("/register", methods=['POST']) def register(): schema = { 'email': { 'type': 'string', 'required': True, 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$' }, 'password': { 'type': 'string', 'required': True, 'min': 6 }, 'first_name': { 'type': 'string', 'required': True, 'min': 2 }, 'last_name': { 'type': 'string', 'required': True, 'min': 2, } } v = Validator(schema, error_handler=CustomErrorHandler) form_data = request.get_json() args = request.args if(v.validate(form_data, schema) == False): return v.errors email = form_data['email'] verification_code = otp(7) try: new_user = User( email= form_data['email'], password = generate_password_hash(form_data['password']), first_name= form_data['first_name'], last_name= form_data['last_name'], verification_code = secret(verification_code), ) db.session.add(new_user) db.session.commit() except SQLAlchemyError as e: # error = str(e.__dict__['orig']) message = str(e) return error({}, message, 400) # return make_response(jsonify({'error': error}), 400) # Send verification email appName = app.config["APP_NAME"].capitalize() email_data = { 'subject': 'Account Verification on ' + appName, 'to': email, 'body': '', 'name': form_data['first_name'], 'callBack': verification_code, 'template': 'verification_email' } send_email.delay(email_data) message = 'Registration Successful, check your email for OTP code to verify your account' return success({}, message, 200)
Enter fullscreen mode Exit fullscreen mode

For validation I used cerberus. Cerberus is a python package that makes validation easy, it returns errors in json format when a validation fail. You can also provide custom error message like below.

from cerberus import errors class CustomErrorHandler(errors.BasicErrorHandler): messages = errors.BasicErrorHandler.messages.copy() messages[errors.REGEX_MISMATCH.code] = 'Invalid Email!' messages[errors.REQUIRED_FIELD.code] = '{field} is required!'
Enter fullscreen mode Exit fullscreen mode

The register function validates the data, if successful, we then generate an otp/verification code using the otp(total) function. We then hash the code for storage in database using the secret(code=None) function.

def secret(code=None): if not code: code = str(datetime.utcnow()) + otp(5) return hashlib.sha224(code.encode("utf8")).hexdigest() def otp(total): return str(''.join(random.choices(string.ascii_uppercase + string.digits, k=total)))
Enter fullscreen mode Exit fullscreen mode

We hash the user's password using the werkzeug.security generate_password_hash inbuilt function in flask and store the record in database. We are using SQLAlchemy to handle this.

After storing the user details, we schedule email to be sent immediately to the user.

# Send verification email appName = app.config["APP_NAME"].capitalize() email_data = { 'subject': 'Account Verification on ' + appName, 'to': email, 'body': '', 'name': form_data['first_name'], 'callBack': verification_code, 'template': 'verification_email' } send_email.delay(email_data)
Enter fullscreen mode Exit fullscreen mode

Then, we return a response to the user

message = 'Registration Successful, check your email for OTP code to verify your account' return success({}, message, 200)
Enter fullscreen mode Exit fullscreen mode

To handle responses, I created a success and error functions under services/api_response.py

from flask import make_response, jsonify def success(data, message: str=None, code: int = 200): data['status'] = 'Success' data['message'] = message data['success'] = True return make_response(jsonify(data), code) def error(data, message: str, code: int): data['status'] = 'Error' data['message'] = message data['success'] = False return make_response(jsonify(data), code)
Enter fullscreen mode Exit fullscreen mode

We have other functions to handle email verification with OTP, login and refresh_token.

Verify account 

@auth.route("/verify", methods=['POST']) def verifyAccount(): schema = { 'otp': { 'type': 'string', 'required': True, 'min': 6 }, } v = Validator(schema, error_handler=CustomErrorHandler) form_data = request.get_json() if(v.validate(form_data, schema) == False): return v.errors otp = form_data['otp'] hash_otp = secret(otp) user = User.query.filter_by(verification_code = hash_otp).first() if not user: message = 'Failed to verify account, Invalid OTP Code' return error({},message, 401) user.verification_code = None user.is_verified = True db.session.commit() message = 'Verification Successful! Login to your account' return success({}, message, 200)
Enter fullscreen mode Exit fullscreen mode

Login user

This function handles POST request and validates the user's email and password. We first check if a record with the user's email exists, if not, we return an error response. If user's email exists, we check_password_hash function to check if the password supplied is valid.
If the password is valid, we call the authenticated function to generate and store a refresh token, generate a JWT token and the return a response with both tokens to the user.

@auth.route("/login", methods=['POST']) def login(): schema = { 'email': { 'type': 'string', 'required': True, 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$' }, 'password': { 'type': 'string', 'required': True, 'min': 6 }, } v = Validator(schema, error_handler=CustomErrorHandler) form_data = request.get_json() user = User.query.filter_by(email = form_data['email']).first() if not user: message = 'Login failed! Invalid account.' return error({}, message, 401) if not check_password_hash(user.password, form_data['password']): message = 'Login failed! Invalid password.' return error({}, message, 401) return authenticated(user) def authenticated(user: User): refresh_token = secret() try: refreshToken = RefreshToken( user_id = user.id, token = refresh_token, expired_at = dt.utcnow() + timedelta(minutes = int(app.config['REFRESH_TOKEN_DURATION'])) ) db.session.add(refreshToken) db.session.commit() except SQLAlchemyError as e: # error = str(e.__dict__['orig']) message = str(e) return error({}, message, 400) # del user['password'] user_schema = UserSchema() data = { "token": jwtEncode(user), "refresh_token": refresh_token, "user": user_schema.dump(user) } message = "Login Successful, Welcome Back" return success(data, message, 200)
Enter fullscreen mode Exit fullscreen mode

The response with the token will look like this:

{ "expired_at": "Wed, 10 Jan 2024 16:43:48 GMT", "message": "Login Successful, Welcome Back", "refresh_token": "a5b5ehghghjk8truur9kj4f999bf6c01d34892df768", "status": "Success", "success": true, "token": "eyK0eCAiOiJDhQiLCJhbGciOI1NiJ9.eyJzdWIiOjEsImlhdCI6MTY3MzM2OTAyOCwiZXhwIjoxNjczOTczODI4fQ.a6fn7z8v9K5EmqZO7-J8VkY2u_Kdffh8aOVuWjTH138", "user": { "avatar": null, "country": { "code": "NG", "id": 158, "name": "Nigeria" }, "country_id": 158, "created_at": "2022-12-09T16:00:48", "email": "user@example.com", "first_name": "John", "id": 145, "is_verified": true, "last_name": "Doe", "updated_at": "2022-12-10T12:17:45" } }
Enter fullscreen mode Exit fullscreen mode

The refresh token is useful for keeping a user logged without requesting for login information every time. Once JWT token is expired a user can use the refresh token to request a new JWT token. This can be handled by the frontend without asking the user for login details.

@auth.route("/refresh", methods=['POST']) def refreshToken(): schema = { 'refresh_token': { 'type': 'string', 'required': True, }, } v = Validator(schema, error_handler=CustomErrorHandler) form_data = request.get_json() now = dt.utcnow() refresh_token = RefreshToken.query.filter(token == form_data['refresh_token'], expired_at >= now).first() if not refresh_token: message = "Token expired, please login" return error({}, message, 401) user = User.query.filter_by(id = refresh_token.user_id).first() if not user: message = "Invalid User" return error({}, message, 403) data = { "token": jwtEncode(user), "id": user.id } message = "Token Successfully refreshed" return success(data, message, 200)
Enter fullscreen mode Exit fullscreen mode

You can access the full source code on GitHub HERE

Step 4: Install and Run the application

You can run the application by executing the following on your terminal

flask --app flaskauth run
Enter fullscreen mode Exit fullscreen mode

Ensure the virtual environment is active and you're on the root project folder when you run this.
Alternatively, you don't need to be in the root project folder to run the command if you installed the application using the command below

$ pip install -e .
Enter fullscreen mode Exit fullscreen mode

To ensure email is delivering set up SMTP crediential in the .env file

APP_NAME=flaskauth FLASK_APP=flaskauth FLASK_DEBUG=True FLASK_TESTING=False SQLALCHEMY_DATABASE_URI=mysql://root:@localhost/flaskauth_db?charset=utf8mb4 MAIL_SERVER='smtp.mailtrap.io' MAIL_USERNAME= MAIL_PASSWORD= MAIL_PORT=2525 MAIL_USE_TLS=True MAIL_USE_SSL=False MAIL_DEFAULT_SENDER='info@flaskauth.app' MAIL_DEBUG=True SERVER_NAME= AWS_SECRET_KEY= AWS_KEY_ID= AWS_BUCKET= JWT_DURATION=10080 REFRESH_TOKEN_DURATION=525600
Enter fullscreen mode Exit fullscreen mode

Also, run the celery application to handle queue and email sending

celery -A flaskauth.celery worker --loglevel=INFO
Enter fullscreen mode Exit fullscreen mode

Step 5: Run Database Migration and Seeding

We will use alembic package to handle migration. Alembic is already installed as part of our requirements. Also, see the source code for the migration files. We will run the following to migrate.

alembic upgrade head
Enter fullscreen mode Exit fullscreen mode

You can use alembic to generate database migrations. For example, to create users table run

alembic revision -m "create users table"
Enter fullscreen mode Exit fullscreen mode

This will generate a migration file, similar to the following:

"""create users table Revision ID: 9d4a5cb3f558 Revises: 5b9768c1b705 Create Date: 2023-01-10 18:34:09.608403 """ from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '9d4a5cb3f558' down_revision = '5b9768c1b705' branch_labels = None depends_on = None def upgrade() -> None: pass def downgrade() -> None: pass
Enter fullscreen mode Exit fullscreen mode

You can then edit it to the following:

"""create users table Revision ID: 9d4a5cb3f558 Revises: None Create Date: 2023-01-10 18:34:09.608403 """ from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '9d4a5cb3f558' down_revision = None branch_labels = None depends_on = None def upgrade() -> None: op.create_table('users', sa.Column('id', sa.BigInteger(), nullable=False), sa.Column('email', sa.String(length=120), nullable=False), sa.Column('password', sa.String(length=200), nullable=True), sa.Column('first_name', sa.String(length=200), nullable=False), sa.Column('last_name', sa.String(length=200), nullable=False), sa.Column('avatar', sa.String(length=250), nullable=True), sa.Column('country_id', sa.Integer(), nullable=True), sa.Column('is_verified', sa.Boolean(), nullable=False), sa.Column('verification_code', sa.String(length=200), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=False), sa.Column('updated_at', sa.DateTime(), nullable=True), sa.Column('deleted_at', sa.DateTime(), nullable=True), sa.ForeignKeyConstraint(['country_id'], ['countries.id'], onupdate='CASCADE', ondelete='SET NULL'), sa.PrimaryKeyConstraint('id'), sa.UniqueConstraint('email') ) # ### end Alembic commands ### def downgrade() -> None: op.drop_table('users') # ### end Alembic commands ###
Enter fullscreen mode Exit fullscreen mode

You can also auto generate migration files from your models using the following:

alembic revision --autogenerate
Enter fullscreen mode Exit fullscreen mode

Note: For auto generation to work, you have to import your app context into the alembic env.py file. See source code for example.

Conclusion

After completing the database migration, you can go ahead and test the app buy sending requests to http://localhost:5000/auth/register using Postman, remember to supply all the necessary data, example below:

{ "email": "ade@example.com", "password": "password", "first_name": "Ade", "last_name": "Emeka" "country": "NG" }
Enter fullscreen mode Exit fullscreen mode

Let me know your thoughts and feedback below.

Top comments (1)

Collapse
 
rudra0x01 profile image
Rudra Sarkar •

well written, write some devops projects