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


Test the Application using Postman

Popular posts from this blog

Spring boot video streaming example-HTML5

Learn Java 8 streams with an example - print odd/even numbers from Array and List

Spring Boot + Mockito simple application with 100% code coverage

Spring Boot + OpenCSV Export Data to CSV Example

Custom Exception Handling in Quarkus REST API

DataTable-Pagination example with Spring boot, jQuery and ajax

Registration and Login with Spring Boot + Spring Security + Thymeleaf

Node JS mini projects with source code - free download

Spring boot web project free download:User Registration System

Java - Blowfish Encryption and decryption Example