Python Flask RESTful API for MySQL CRUD

Posted on 13th August 2020

In this tutorial, you build a simple web application using Python Flask that can read from and write to a MySQL database. The aim is to learn three main aspects of Flask.

  • How to create a web application.
  • How to build REST API using Python Flask.
  • How to perform CRUD operations on MySQL database using Flask.

This tutorial assumes you are familiar with Python Flask. If not, read Getting started with Python Flask on Windows and Linux to learn how to build a basic Hello world app.

Pre-requisites

The following software must be installed.

  • Python 3.x
  • Flask
  • MySQL

Creating Table in MySQL

For the purpose of this demo app that we are building, we also require a table in MySQL to which we can insert new data, query existing records and also delete and update records. Here is the SQL statement to create a table named otg_demo_users. You can run these from any MySQL client.

CREATE TABLE otg_demo_users (
  id INT NOT NULL AUTO_INCREMENT,
  name VARCHAR(45) NOT NULL,
  age INT NULL,
  city VARCHAR(45) NULL,
PRIMARY KEY (`id`));

We can also insert a few rows in this table.

INSERT INTO otg_demo_users(name, age, city) VALUES('Jake', 21, 'London');
INSERT INTO otg_demo_users(name, age, city) VALUES('Mathew', 24, 'Sydney');
INSERT INTO otg_demo_users(name, age, city) VALUES('Bob', 35, 'Newyork');
commit;

Application setup

In the next step, we setup the application directory, virtual environment and install the dependency packages. Run the following commands

On Windows,

PS C:\> mkdir FlaskCRUD
PS C:\> cd FlaskCRUD
PS C:\FlaskCRUD> py -3 -m venv env
(env) PS C:\FlaskCRUD> .\env\Scripts\Activate.ps1
(env) PS C:\FlaskCRUD> pip install flask
(env) PS C:\FlaskCRUD> pip install flask-mysql
(env) PS C:\FlaskCRUD> pip install flask-restful
(env) PS C:\FlaskCRUD> $env:FLASK_APP = "app.py"
(env) PS C:\FlaskCRUD> $env:FLASK_ENV = "development"

Application Source

Now let's jump into the fun part, coding.

  1. Import flask, jsonify and request from Flask library. jsonify is for sending server responses in JSON format and request is for getting the incomming request data.

    To access MYSQL database, we use the flask extension Flask-MySQL.

    We also need another flask extension, Flask-RESTful, which enables building REST APIs easily.

  2. Create instances of Flask, MySQL and Flask-RESTful.

  3. Load database connection details to the config attribute of Flask object and initialize the Flask-MySQL extension.

from flask import Flask, jsonify, request
from flaskext.mysql import MySQL
from flask_restful import Resource, Api

#Create an instance of Flask
app = Flask(__name__)

#Create an instance of MySQL
mysql = MySQL()

#Create an instance of Flask RESTful API
api = Api(app)

#Set database credentials in config.
app.config['MYSQL_DATABASE_USER'] = 'user_name'
app.config['MYSQL_DATABASE_PASSWORD'] = 'password'
app.config['MYSQL_DATABASE_DB'] = 'database_name'
app.config['MYSQL_DATABASE_HOST'] = 'server_name'

#Initialize the MySQL extension
mysql.init_app(app)


#Get All Users, or Create a new user
class UserList(Resource):
    def get(self):
        try:
            conn = mysql.connect()
            cursor = conn.cursor()
            cursor.execute("""select * from otg_demo_users""")
            rows = cursor.fetchall()
            return jsonify(rows)
        except Exception as e:
            print(e)
        finally:
            cursor.close()
            conn.close()

    def post(self):
        try:
            conn = mysql.connect()
            cursor = conn.cursor()
            _name = request.form['name']
            _age = request.form['age']
            _city = request.form['city']
            insert_user_cmd = """INSERT INTO otg_demo_users(name, age, city) 
                                VALUES(%s, %s, %s)"""
            cursor.execute(insert_user_cmd, (_name, _age, _city))
            conn.commit()
            response = jsonify(message='User added successfully.', id=cursor.lastrowid)
            #response.data = cursor.lastrowid
            response.status_code = 200
        except Exception as e:
            print(e)
            response = jsonify('Failed to add user.')         
            response.status_code = 400 
        finally:
            cursor.close()
            conn.close()
            return(response)
            
#Get a user by id, update or delete user
class User(Resource):
    def get(self, user_id):
        try:
            conn = mysql.connect()
            cursor = conn.cursor()
            cursor.execute('select * from otg_demo_users where id = %s',user_id)
            rows = cursor.fetchall()
            return jsonify(rows)
        except Exception as e:
            print(e)
        finally:
            cursor.close()
            conn.close()

    def put(self, user_id):
        try:
            conn = mysql.connect()
            cursor = conn.cursor()
            _name = request.form['name']
            _age = request.form['age']
            _city = request.form['city']
            update_user_cmd = """update otg_demo_users 
                                 set name=%s, age=%s, city=%s
                                 where id=%s"""
            cursor.execute(update_user_cmd, (_name, _age, _city, user_id))
            conn.commit()
            response = jsonify('User updated successfully.')
            response.status_code = 200
        except Exception as e:
            print(e)
            response = jsonify('Failed to update user.')         
            response.status_code = 400
        finally:
            cursor.close()
            conn.close()    
            return(response)       

    def delete(self, user_id):
        try:
            conn = mysql.connect()
            cursor = conn.cursor()
            cursor.execute('delete from otg_demo_users where id = %s',user_id)
            conn.commit()
            response = jsonify('User deleted successfully.')
            response.status_code = 200
        except Exception as e:
            print(e)
            response = jsonify('Failed to delete user.')         
            response.status_code = 400
        finally:
            cursor.close()
            conn.close()    
            return(response)       

#API resource routes
api.add_resource(UserList, '/users', endpoint='users')
api.add_resource(User, '/user/<int:user_id>', endpoint='user')

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


Post a comment

Comments

Nothing yet..be the first to share wisdom.