Create a CRUD Restful Service API using Flask + Mysql [in 7 minutes!]

In this article, we will learn how to build Simple Restful API with flask and Mysql that have capabilities to create, read, update, and delete data from the database.

Flask being a microframework provides the flexibility of the data source for applications and also provides library support for interacting with different kinds of data sources. There are libraries to connect to SQL- and NoSQL-based databases in Flask.



Creating a Flask Application with SQLAlchemy

Flask-SQLAlchemy is an extension for flask which adds support for SQLAlchemy to the application. SQLAlchemy is a Python toolkit and Object Relational Mapper that provides access to the SQL database using Python. SQLAlchemy comes with enterprise-level persistence patterns and efficient and high performing database access. 

Flask-SQLAlchemy provides support for the following SQL-based database engines given the appropriate DBAPI driver is installed:

  • PostgreSQL
  • MySQL
  • Oracle
  • SQLite
  • Microsoft SQL Server
  • Firebird Sybase
We’ll be using MySQL as the database engine in our application, so let’s get started with installing SQLAlchemy and start setting up our application for our CRUD based Restful service.


Note: While working on product proper folder structure should be maintained which I will be explain in further posts.

SETTING UP ENVIRONMENT


Let’s create a new directory called flaskdbexample, create a virtual environment, and then install flask-sqlalchemy.

mkdir flaskdbexample  
cd flaskdbexample

Now, create a virtual environment inside the directory using the following command:

virtualenv venv

As discussed earlier, we can activate the virtual environment using the following command:

venv/Script/activate

Once the virtual environment is activated, let’s install flask-sqlalchemy.
Flask and Flask-SQLAlchemy can be installed using PIP with the following command.

(venv)> pip install flask flask-sqlalchemy

Now, let’s install PyMySQL to enable MySQL connection with Flask-SQLAlchemy.


(venv)> pip install pymysql

Let’s start by creating app.py which will contain the code for our application. After creating the file, we’ll initiate the Flask application.


from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields

On this part, we import all modules that needed by our application. We import Flask to create an instance of a web application, request to get request data, jsonify to turns the JSON output into a Response object with the application/json mimetype, SQAlchemy from flask_sqlalchemy to accessing database, and Marshmallow from flask_marshmallow to serialized object.


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI']='mysql+pymysql://<mysql_username>:<mysql_password>@<mysql_host>:<mysql_port>/<mysql_db>'
db = SQLAlchemy(app)

we configure the SQLAlchemy database URI to use our MySQL DB URI, and then we create an object of SQLAlchemy named as db, which will handle our ORM-related activities.

CREATING DATABASE

We’ll now create a product database application that will provide RESTful CRUD APIs. All the products will be stored in a table titled "products".

After the declared DB object, add the following lines of code to declare a class as Product which will hold the schema for the products table:


###Models####
class Product(db.Model):
    __tablename__ = "products"
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(20))
    productDescription = db.Column(db.String(100))
    productBrand = db.Column(db.String(20))
    price = db.Column(db.Integer)

    def create(self):
      db.session.add(self)
      db.session.commit()
      return self
    def __init__(self,title,productDescription,productBrand,price):
        self.title = title
        self.productDescription = productDescription
        self.productBrand = productBrand
        self.price = price
    def __repr__(self):
        return '' % self.id
db.create_all()

we have created a model titled "Product" which has five fields— ID is a self-generated and auto-incremented integer which will serve as a primary key. “db.create_all()” which instructs the application to create all the tables and database specified in the application.


class ProductSchema(ModelSchema):
    class Meta(ModelSchema.Meta):
        model = Product
        sqla_session = db.session
    id = fields.Number(dump_only=True)
    title = fields.String(required=True)
    productDescription = fields.String(required=True)
    productBrand = fields.String(required=True)
    price = fields.Number(required=True)

The preceding code maps the variable attribute to field objects, and in Meta, we define the model to relate to our schema. So this should help us return JSON from SQLAlchemy.

DESIGNING ENDPOINTS FOR CRUD

After setting up our model and return schema, we can jump to creating our endpoints. Let’s create our first GET /products endpoint to return all the

THE GET METHOD

@app.route('/products', methods = ['GET'])
def index():
    get_products = Product.query.all()
    product_schema = ProductSchema(many=True)
    products = product_schema.dump(get_products)
    return make_response(jsonify({"product": products}))
In this method, we are fetching all the products in the DB, dumping it in the ProductSchema, and returning the result in JSON.

If you start the application and hit the endpoint now, it will return an empty array since we haven’t added anything in the DB yet, but let’s go ahead and try the endpoint. Btw I have one record added

To run the application

Flask run 


THE POST METHOD:


@app.route('/products', methods = ['POST'])
def create_product():
    data = request.get_json()
    product_schema = ProductSchema()
    product = product_schema.load(data)
    result = product_schema.dump(product.create())
    return make_response(jsonify({"product": result}),200)





THE UPDATE (PUT) METHOD:


@app.route('/products/<id>', methods = ['PUT'])
def update_product_by_id(id):
    data = request.get_json()
    get_product = Product.query.get(id)
    if data.get('title'):
        get_product.title = data['title']
    if data.get('productDescription'):
        get_product.productDescription = data['productDescription']
    if data.get('productBrand'):
        get_product.productBrand = data['productBrand']
    if data.get('price'):
        get_product.price= data['price']    
    db.session.add(get_product)
    db.session.commit()
    product_schema = ProductSchema(only=['id', 'title', 'productDescription','productBrand','price'])
    product = product_schema.dump(get_product)
    return make_response(jsonify({"product": product}))




THE DELETE  BY ID METHOD :


@app.route('/products/<id>', methods = ['DELETE'])
def delete_product_by_id(id):
    get_product = Product.query.get(id)
    db.session.delete(get_product)
    db.session.commit()
    return make_response("",204)
 

Complete Code for Testing:


from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://<username>:<pass>@localhost:3306/<DB>'
db = SQLAlchemy(app)

###Models####
class Product(db.Model):
    __tablename__ = "products"
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(20))
    productDescription = db.Column(db.String(100))
    productBrand = db.Column(db.String(20))
    price = db.Column(db.Integer)

    def create(self):
      db.session.add(self)
      db.session.commit()
      return self
    def __init__(self,title,productDescription,productBrand,price):
        self.title = title
        self.productDescription = productDescription
        self.productBrand = productBrand
        self.price = price
    def __repr__(self):
        return '' % self.id
db.create_all()
class ProductSchema(ModelSchema):
    class Meta(ModelSchema.Meta):
        model = Product
        sqla_session = db.session
    id = fields.Number(dump_only=True)
    title = fields.String(required=True)
    productDescription = fields.String(required=True)
    productBrand = fields.String(required=True)
    price = fields.Number(required=True)

@app.route('/products', methods = ['GET'])
def index():
    get_products = Product.query.all()
    product_schema = ProductSchema(many=True)
    products = product_schema.dump(get_products)
    return make_response(jsonify({"product": products}))
@app.route('/products/<id>', methods = ['GET'])
def get_product_by_id(id):
    get_product = Product.query.get(id)
    product_schema = ProductSchema()
    product = product_schema.dump(get_product)
    return make_response(jsonify({"product": product}))
@app.route('/products/<id>', methods = ['PUT'])
def update_product_by_id(id):
    data = request.get_json()
    get_product = Product.query.get(id)
    if data.get('title'):
        get_product.title = data['title']
    if data.get('productDescription'):
        get_product.productDescription = data['productDescription']
    if data.get('productBrand'):
        get_product.productBrand = data['productBrand']
    if data.get('price'):
        get_product.price= data['price']    
    db.session.add(get_product)
    db.session.commit()
    product_schema = ProductSchema(only=['id', 'title', 'productDescription','productBrand','price'])
    product = product_schema.dump(get_product)
    return make_response(jsonify({"product": product}))
@app.route('/products/<id>', methods = ['DELETE'])
def delete_product_by_id(id):
    get_product = Product.query.get(id)
    db.session.delete(get_product)
    db.session.commit()
    return make_response("",204)
@app.route('/products', methods = ['POST'])
def create_product():
    data = request.get_json()
    product_schema = ProductSchema()
    product = product_schema.load(data)
    result = product_schema.dump(product.create())
    return make_response(jsonify({"product": result}),200)
if __name__ == "__main__":
    app.run(debug=True)




Conclusion:
So, we have now created and tested our sample Flask-MySQL CRUD application. We’ll go over complex object relationships using Flask-SQLAlchemy in the later post, and next we’ll create a similar Flask CRUD application.

Happy Coding!



Hey I'm Venkat
Developer, Blogger, Thinker and Data scientist. nintyzeros [at] gmail.com I love the Data and Problem - An Indian Lives in US .If you have any question do reach me out via below social media