Dubai Silicon Oasis, Dubai, UAE

Flask Authentication

A Flask API with Authentication and Credit Management

There is a ton of sample codes out there on StackOverflow or GitHub that help to create Flask API and allow authentication for this API.

I’m not very good in embedding part of the code in a post, so I will just leave the code here and the post to be edited later.

In this post, I will cover a way to create the Flask API, allow authentication on this API with username and token saved in MySQL Database.

I will also show how I managed to control the registered users’ credit so that each user has a credit score to use the API. This credit can be reset each month through a separate function not showing in the sample code.

I will also show in the code how to add and remove a background task and set a run time for it.

The sample code can be found on GitHub:

from flask import Flask, jsonify, abort, request, Response
from flask_login import LoginManager, UserMixin, login_required
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
import mysql.connector
import json
import os
import signal
import logging
import datetime
sched = BackgroundScheduler()
# I'm disabling Logging as other parts of the code need to have clear log without too much of nothing to do logs
logging.getLogger('apscheduler.scheduler').propagate = False
def job():
# The ID here is mandatory to remove this function from Background once needed
sched.add_job(job, 'interval', seconds=10, id='main')
# Flask def
app = Flask(__name__)
# app.config['ENV'] = 'development'
app.config['ENV'] = 'deployment'
# app.config['DEBUG'] = True
# app.config['TESTING'] = True
app.secret_key = 'KEYSECRET' # Change this!
login_manager = LoginManager()
def sendMail(subject, body, attach, email):
subject = 'LOCAL - ' + subject
if email == None:
to = ''
to = email
yag = yagmail.SMTP("", 'password')
subject= subject,
def db_check():
mydb = mysql.connector.connect(host="localhost",user="Manager",password="ManagerPWD",database = "DBNAME")
mycursor = mydb.cursor(dictionary=True)
rows = mycursor.fetchall()
newdata = [{dic['USERNAME']: (dic['TOKEN'], dic['CREDIT'])} for dic in rows]
diction_user_token = {}
for item in newdata:
for key, value in item.items():
diction_user_token[key] = value
user_database = diction_user_token
return user_database
class User(UserMixin):
# Load your users from DB here if you wish to have an instant from it, otherwise put in the get function below
# user_database = db_check()
# Sample of the returned data
# user_database = {"melbasyouni": ("p50Bux2fe1JCLkkkVYqt6r,G", 10000000),"JaneDoe": ("Jane", 10000000)}
def __init__(self, username, password): = username
self.password = password
def get(cls,id):
user_database = db_check()
# Sample of the returned data
# user_database = {"melbasyouni": ("p50Bux2fe1JCLkkkVYqt6r,G", 10000000),"JaneDoe": ("Jane", 10000000)}
return user_database.get(id)
# API Request should have ?token=username:password at the end of the URL
def load_user(request):
global credit_score
token = request.headers.get('Authorization')
if token is None:
token = request.args.get('token')
if token is not None:
username,password = token.split(":")
user_entry = User.get(username)
if (user_entry is not None):
user = User(username,user_entry)
if (user.password[0] == password):
# Check if user has credit and update global variable
if (user.password[1] > 0):
new_value = user.password[1] - 1
credit_score = True
# Update the Database with the new credit for this user
sql_queries.users_credit_update_API(user.password[0], new_value)
return user
credit_score = False
return user
return None
def unauthorized_callback():
return (jsonify({'Status': 'Access Not Authorized'}), 201)
def before():
print("This is executed BEFORE each request.")
def warmup():
# Handle your warmup logic here, e.g. set up a database connection pool
return '', 200, {}
@app.route('/api/v1.0/banks_data/<bank>/', methods = ['GET'])
def banks_data(bank):
# Check if user has credit
if credit_score == True:
if bank != None:
# Return the needed Data
return (jsonify({'Data': 'Hope you get the data'}), 201)
return (jsonify({'Status': 'Requested Bank Code can not be None'}), 201)
# Check if user has credit failed
return (jsonify({'Status': 'No enough credit for this account'}), 201)
@app.route('/stopServer', methods=['GET'])
def stopServer():
try:"Stopping Flask API")
subject = 'Flask API Logic Stopped'
body = 'Flask API Logic Stopped with embedded functions.'
# Remove Background Function before stopping the Flask API
if sched is not None:
# Good Idea to send Mail in case API stopped
sendMail(subject, body, None, None)
os.kill(os.getpid(), signal.SIGINT)
return jsonify({ "success": True, "message": "Server is Stopping..." })
except Exception as err:"Something went wrong with Flask API, error is '" + str(err) + "'")
subject = 'Flask API Logic Stopping Failed'
body = "Something went wrong with Flask API, error is '" + str(err) + "'"
# Good Idea to send Mail in case API failed to stop
sendMail(subject, body, None, None)
return (jsonify({'Status': 'Something went wrong when stopping server'}), 201)
if __name__ == '__main__':
sched.start()'', port=5000)
# In case SSL is needed, an adhoc ssl context can be added as a parameter to the
# ssl_context='adhoc'
except Exception as err:"Something went wrong with Flask API, error is '" + str(err) + "'")
subject = 'Flask API Logic Failed'
body = "Something went wrong with Flask API, error is '" + str(err) + "'"
# It's a good idea to send Mail in case the App Failed to run => A Cloud Practice I'm using
sendMail(subject, body, None, None)
view raw hosted with ❤ by GitHub
Privacy Preferences
When you visit our website, it may store information through your browser from specific services, usually in form of cookies. Here you can change your privacy preferences. Please note that blocking some types of cookies may impact your experience on our website and the services we offer.