TheLazy.Dev
Portfolio GitHub

Blog

python |flask |api |mysql

Building a User Authentication API using Python Flask and MySQL

andychukse

Andy

Posted on January 10, 2023

Building a User Authentication API using Python Flask and 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.

💖 💪 🙅 🚩
andychukse
Andy

Posted on January 10, 2023

Join Our Newsletter. No Spam, Only the good stuff.

Sign up to receive the latest update from our blog.

Related

Building a User Authentication API using Python Flask and MySQL
python Building a User Authentication API using Python Flask and MySQL

January 10, 2023