Building a User Authentication API using Python Flask and MySQL
Tags: python, flask, api, mysql
Published: 2023-01-10 18:03:24

Recently, I decided to learn Python, as part of learning I built a remote jobs ([remote4africa](https://remote4africa.com)) 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 
```
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](https://github.com/andychukse/flaskauth)).

**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
            ...
```

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"
]

```

**setup.py**

```
from setuptools import setup

setup(
    name='yourapplication',
    packages=['yourapplication'],
    include_package_data=True,
    install_requires=[
        'flask',
    ],
    py_modules=['config']
)
```
You can then install your application so it is importable:

```
$ pip install -e .
```
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
```

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
    ...
```


## 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
```

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)
```

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)
```

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)
```
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)
```

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)
```
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"})
```
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)
```

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!'
```

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)))
```

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)
```

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)
```
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)
```

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)
```

### 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)
```

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"
    }
}
```

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)
```

**You can access the full source code on [GitHub HERE](https://github.com/andychukse/flaskauth)**


## Step 4: Install and Run the application

You can run the application by executing the following on your terminal
```
flask --app flaskauth run
```
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 .
```
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
```

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

```
celery -A flaskauth.celery worker --loglevel=INFO
```

## 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
```

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

```
alembic revision -m "create users table"
```
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
```
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 ###
```

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

```
alembic revision --autogenerate
```

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"
}
```

Let me know your thoughts and feedback below.