Flask – Simple Restful API (mysql.connector)

– Create Database : db_simple

– Install library :

pip install mysql-connector-python

– Create file : database.py

import mysql.connector

def database():
    db = mysql.connector.connect(
        host="localhost",
        user="root",
        password="",
        database="db_simple"
    )
    return db

– Create file : migration.py

import mysql.connector
from database import database

db = database()
cursor = db.cursor()

def create_table_users():
    try:
        sql = """CREATE TABLE users (
              id INT AUTO_INCREMENT PRIMARY KEY,
              name VARCHAR(255),
              email VARCHAR(255),
              password VARCHAR(255)
              )
              """
        cursor.execute(sql)
        print("- Users Table created successfully .....")
    except:
        print("- Users Table Already Exist ! .....")


# Execute :
if db.is_connected():
  print("- Connecting to Database successfully .....")
  create_table_users()

– Run migration table :

python3 migration.py

– Create file : app.py

from flask import Flask, request
import mysql.connector
from response import *
from database import database

app = Flask(__name__)

db = database()
cursor = db.cursor()

@app.route('/')
def index():
    return "index page"

@app.route('/users', methods=['GET','POST'])
def users():
    if request.method == 'POST':

        name = request.form.get('name')
        email = request.form.get('email')
        password = request.form.get('password')

        input = [{
            'name': name,
            'email': email,
            'password': password
        }]

        sql = "INSERT INTO users (name, email, password) VALUES (%s, %s, %s)"
        val = (name, email, password)
        cursor.execute(sql, val)
        db.commit()
        return success(input, "success")

    else:

        sql = "SELECT * FROM users"
        cursor.execute(sql)
        users = cursor.fetchall()
        data = array_users(users)
        return success(data, "success")

@app.route('/user/<int:id>', methods=['GET','PUT','DELETE'])
def user(id):
    if request.method == 'GET':

        try:
            sql = "SELECT * FROM users WHERE id = " + str(id)
            cursor.execute(sql)
            users = cursor.fetchone()
            data = obj_users(users)
            return success(data, "success")
        except:
            return badRequest([], 'Data Not Found')

    elif request.method == 'PUT':

        name = request.form.get('name')
        email = request.form.get('email')
        password = request.form.get('password')

        input = [{
            'name': name,
            'email': email,
            'password': password
        }]

        sql = "UPDATE users SET name = %s, email = %s, password = %s WHERE id = %s"
        val = (name, email, password, id)
        cursor.execute(sql, val)
        db.commit()
        return success(input, "success")

    elif request.method == 'DELETE':

        sql = "DELETE FROM users WHERE id = " + str(id)
        cursor.execute(sql)
        db.commit()
        return success('', 'Success Deleted')



if __name__ == '__main__':
    app.run(debug=True)


# set FLASK_APP=app.py
# set FLASK_ENV=development
# flask run

– Create file : response.py

from flask import jsonify, make_response

# RESPONSE SUCCESS
def success(values, message):
    res = {
        'data' : values,
        'message': message
    }

    return make_response(jsonify(res)), 200

# RESPONSE BAD REQUEST
def badRequest(values, message):
    res = {
        'data' : values,
        'message': message
    }

    return make_response(jsonify(res)), 400

# ARRAY TABLE USERS
def array_users(datas):
    values = []

    for data in datas:
        dict = {}
        dict['id'] = data[0]
        dict['name'] = data[1]
        dict['email'] = data[2]
        dict['password'] = data[3]

        values.append(dict)

    return values

# SINGLE OBJECT TABLE USERS
def obj_users(data):

    dict = {}
    dict['id'] = data[0]
    dict['name'] = data[1]
    dict['email'] = data[2]
    dict['password'] = data[3]

    return dict

– Run app :

set FLASK_APP=app.py
set FLASK_ENV=development
flask run

Leave a Reply

Your email address will not be published. Required fields are marked *