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.
Test the Application using Postman
Add Employee:
Comments
Post a Comment