Python-Create a CRUD Restful Service API using FLASK and MYSQL

Hello everyone, Today we will learn how to build Restful API with FLASK and MYSQL that have the capabilities to create, read, update, and delete data from the MYSQL database. You can download the source code from our Github repository. Link is provided at the end of this post.

More Related Topics,

Technologies used:

Python 3
Flask library
pymysql library
flask-sqlalchemy library

Please go through the following steps in order to implement Python Restful API CRUD example using Flask and MySQL:

Step 1. Create MySQL database table – 'employee' with the following structure.

CREATE TABLE IF NOT EXISTS `employee` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `uname` varchar(45) COLLATE utf8_unicode_ci DEFAULT NULL,
  `email` varchar(45) COLLATE utf8_unicode_ci DEFAULT NULL,
  `salary` varchar(255) COLLATE utf8_unicode_ci DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci AUTO_INCREMENT
=2343 ;

Step 2. Create the below app.py script(py is the extension to indicate Python script) where we import the flask module.

app.py:

from flask import Flask
app = Flask(__name__)

Step 3.create the below db_config.py Python script o setup the MySQL database configurations for connecting to the database.

db_config.py:

from app import app
from flaskext.mysql import MySQL
mysql = MySQL()
# MySQL configurations
app.config['MYSQL_DATABASE_USER'] = 'root'
app.config['MYSQL_DATABASE_PASSWORD'] = ''
app.config['MYSQL_DATABASE_DB'] = 'knf'
app.config['MYSQL_DATABASE_HOST'] = 'localhost'
mysql.init_app(app) 

Step 4.Next we need 'resttcontroller.py' script. This script is the perfect instance of Python REST API CRUD Example using Flask and MySQL. It defines all REST URIs for performing CRUD operations

resttcontroller.py:

import pymysql
from app import app
from db_config import mysql
from flask import jsonify
from flask import flash, request

@app.route('/addEmployee', methods=['POST'])
def add_employee():
 try:
  _json = request.json
  _id =_json['id']
  _name = _json['name']
  _email = _json['email']
  _salary =_json['salary']
  # validate the received values
  if _id and _name and _email and _salary and request.method == 'POST':
   # save edits
   sql = "INSERT INTO employee(id, uname, email,salary) VALUES(%s, %s, %s ,  %s)"
   data = (_id, _name, _email,_salary)
   conn = mysql.connect()
   cursor = conn.cursor()
   cursor.execute(sql, data)
   conn.commit()
   resp = jsonify('Employee added successfully!')
   resp.status_code = 200
   return resp
  else:
   return not_found()
 except Exception as e:
  print(e)
 finally:
  cursor.close() 
  conn.close()
  
@app.route('/employeeList')
def employees():
 try:
  conn = mysql.connect()
  cursor = conn.cursor(pymysql.cursors.DictCursor)
  cursor.execute("SELECT * FROM employee")
  rows = cursor.fetchall()
  resp = jsonify(rows)
  resp.status_code = 200
  return resp
 except Exception as e:
  print(e)
 finally:
  cursor.close() 
  conn.close()
  
@app.route('/employee/<int:id>')
def employee(id):
 try:
  conn = mysql.connect()
  cursor = conn.cursor(pymysql.cursors.DictCursor)
  cursor.execute("SELECT * FROM employee WHERE id=%s", id)
  row = cursor.fetchone()
  resp = jsonify(row)
  resp.status_code = 200
  return resp
 except Exception as e:
  print(e)
 finally:
  cursor.close() 
  conn.close()

@app.route('/employeeUpdate', methods=['POST'])
def update_employee():
 try:
  _json = request.json
  _id =_json['id']
  _name = _json['name']
  _email = _json['email']
  _salary =_json['salary']  
  # validate the received values
  if _id and _name and _email and _salary and request.method == 'POST':
   # save edits
   sql = "UPDATE employee SET uname=%s, email=%s, salary=%s WHERE id=%s"
   data = (_name, _email, _salary, _id,)
   conn = mysql.connect()
   cursor = conn.cursor()
   cursor.execute(sql, data)
   conn.commit()
   resp = jsonify('Employee updated successfully!')
   resp.status_code = 200
   return resp
  else:
   return not_found()
 except Exception as e:
  print(e)
 finally:
  cursor.close() 
  conn.close()
  
@app.route('/delete/<int:id>')
def delete_employee(id):
 try:
  conn = mysql.connect()
  cursor = conn.cursor()
  cursor.execute("DELETE FROM employee WHERE id=%s", (id,))
  conn.commit()
  resp = jsonify('Employee deleted successfully!')
  resp.status_code = 200
  return resp
 except Exception as e:
  print(e)
 finally:
  cursor.close() 
  conn.close()
  
@app.errorhandler(404)
def not_found(error=None):
    message = {
        'status': 404,
        'message': 'Not Found: ' + request.url,
    }
    resp = jsonify(message)
    resp.status_code = 404

    return resp
  
if __name__ == "__main__":
    app.run()

Run the Application

Now navigate to the directory and execute the command python restcontroller.py as shown in the below image, your server will start on default port 5000.