Flask-RESTful with SQLAlchemy

In the previous article, we’ve created a Fuel Consumption Api which uses dictionary as a database. The biggest disadvantage of this solution is lack of data persistence. To make our application more useful we’ll use SQL database instead of Python dictionary. To achieve this the perfect choice is SQLAlchemy – Python SQL Toolkit and Object Relation Mapper.

Prerequisite

1. Installed Python 3.x
2. Code editor of your choice (e.g. PyCharm, Visual Studio Code)

Project setup

If project from last article is already setup, you can omit steps 1-3. There are new dependencies so step 4 needs to be performed again for new modules.

1. Create new project

mkdir fuel-consumption-api
cd drivers-api

2. Create virtual environment

python3 -m venv flask_venv

3. Activate created venv

source flask_venv/bin/activate

4. Install required dependencies
   a. Manually with use of pip

pip install flask
pip install flask-restful
pip install flask-sqlalchemy
pip install mysql-connector-python

2. Alternatively, you can create a **requirements.txt** file with dependencies:

flask
flask-restful
flask-sqlalchemy
mysql-connector-python

And install them with command:

pip install -r requirements.txt 

Configure connection to database

Code:

from flask import Flask
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)

Code description

1. SQLAlchemy is available in flask_sqlalchemy package.

from flask_sqlalchemy import SQLAlchemy

2. For the project purposes sqlite database is used. Configuration key SQLALCHEMYDATABASEURI needs to be set and SQLAlchemy object needs to be created.

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)

URI variable contains following information

dialect:///absolute\\path\to\fuel.db

While connecting to another type of database URI will contain more information

dialect+driver://username:password@host:port/database

Database model

SQLAlchemy has an ORM (Object-Relational Mapper). To use it database model needs to be created

Code

class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }

Code description

1. Database model is an Python object which inherits from SQLAlchemy.Model class. Class fields are representing the database Columns.

class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

2. In class FuelConsumptionRecord method serialize is defined. The purpose of defining this method is to easily serialize data from Object to return it as a son.

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }

Initialization of Database

Connection with database is configured, model is defined. Now it’s possible to create table.

1. At the beginning table needs to be created:

from fuel_consumption_api import db
db.create_all()

2. Create example data:

from fuel_consumption_api import FuelConsumptionRecord
first_record =  FuelConsumptionRecord(odometer=0, fuelQuantity=0.0)
second_record = FuelConsumptionRecord(odometer=100, fuelQuantity=12.5)
third_record = FuelConsumptionRecord(odometer=110, fuelQuantity=12.5)

3. Put data into the database:

db.session.add(first_record)
db.session.add(second_record)
db.session.add(third_record)
db.session.commit()

4. Check if data is stored in database

FuelConsumptionRecord.query.all()

Retrieve data methods implementation

Code

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return [FuelConsumptionRecord.serialize(record) for record in records]


class FuelConsumption(Resource):
    def get(self, record_id):
        return FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id)))


api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')

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

Code description

1. To retrieve all records stored in database we need simply query.all() on FuelConsumptionRecord class.

records = FuelConsumptionRecord.query.all()

2. Above query returns a list of FuelConsumptionRecord objects. To return them to the user, the objects needs to be serialized.

return [FuelConsumptionRecord.serialize(record) for record in records]

3. To retrieve specific record by record_id field query needs to be filtered.

return FuelConsumptionRecord.serialize(
    FuelConsumptionRecord.query
                         .filter_by(id=record_id)
                         .first_or_404(description='Record with id={} is not available'.format(record_id)))

To return only one element method first_or_404 is called. The advantage of this method is that, in case if query returns zero, 404 http status code will be returned, instead of returning internal server error. Error message can be easily configured with description parameter.

Testing

Let’s check if business requirements for retrieving data are met

1. User is able to retrieve a single record

Query:

curl -X GET http://localhost:5000/recordList

Expected response:

    [
        {
            "id": 1,
            "odometer": 0.0,
            "fuelQuantity": 0.0
        },
        {
            "id": 2,
            "odometer": 100.0,
            "fuelQuantity": 12.5
        },
        {
            "id": 3,
            "odometer": 110.0,
            "fuelQuantity": 12.5
        }
    ]

2. User is able to retrieve a single record

Query:

curl -X GET http://localhost:5000/record/2

Expected response:

    {
        "id": 2,
        "odometer": 100.0,
        "fuelQuantity": 12.5
    }

Record, update and delete methods implementation

Code

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return jsonify([FuelConsumptionRecord.serialize(record) for record in records])

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201


class FuelConsumption(Resource):
    def get(self, record_id):
        return jsonify(FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id))))

    def delete(self, record_id):
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        db.session.delete(record)
        db.session.commit()
        return '', 204

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201

api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')


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

Code description

1. Create new record

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201

To create a new record in database, new FuelConsumptionRecord method is created. Parameters are read from request body. Newly created object needs to be added to SQLAlchemy session and the committed to save in database. To return a newly created object data in response, object needs to be serialized.

2. Update existing record

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201

In case of updating the record as a first step we need to retrieve object from database. Next object fields are overwritten and session is committed.

3. Delete existing record

    def delete(self, record_id):
            record = FuelConsumptionRecord.query.filter_by(id=record_id)\
                .first_or_404(description='Record with id={} is not available'.format(record_id))
            db.session.delete(record)
            db.session.commit()
            return '', 204

To delete record from database, we need to retrieve object from database and then use delete method of SQLAlchemy session to take an action. At the end session needs to be committed.

Testing

1. User is able to save a record

Query:

    curl -X POST \
    http://localhost:5000/recordList \
    -H 'Content-Type: application/json' \
    -d '{
        "odometer": 400,
        "fuelQuantity": 12
    }'

Expected response:

    {
        "id": 4,
        "odometer": 400.0,
        "fuelQuantity": 24.0
    }

Response with 201 http status code (CREATED) is returned

2. User is presented with an error message following sent request without required fields

Query:

    curl -X POST \
    http://localhost:5000/recordList \
    -H 'Content-Type: application/json' \
    -d '{}'

Expected response:

400 BAD REQUEST http response code with body

    {
        "message": {
            "odometer": "odometer is required parameter!",
            "fuelQuantity": "fuelQuantity is required parameter!"
        }
    }

3. User is able to update existing record

Query:

    curl -X PUT \
    http://localhost:5000/record/3 \
    -H 'Content-Type: application/json' \
    -d '{
    "odometer": 120,
    "fuelQuantity": 13.7
    }'

Expected response:

    {
        "id": 3,
        "odometer": 120.0,
        "fuelQuantity": 13.7
    }

4. User is able to delete a single record

Query:

curl -X DELETE http://localhost:5000/record/2

Response:

Response with 204 http status code (NO_CONTENT) is returned

5. To check if above operations are successful, call with GET method /recordList endpoint

Query:

curl -X GET http://localhost:5000/recordList

Expected response:

Valid response body is presented below:

   [
        {
            "id": 1,
            "odometer": 0.0,
            "fuelQuantity": 0.0
        },
        {
            "id": 3,
            "odometer": 120.0,
            "fuelQuantity": 13.7
        },
        {
            "id": 4,
            "odometer": 400.0,
            "fuelQuantity": 24.0
        }
    ]

Statistic methods implementation

Code

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


def calculate_consumption(fuel_quantity, distance):
    return fuel_quantity / distance * 100


def calculate_distance(start_distance, end_distance):
    return end_distance - start_distance


def get_record_by_order_desc(records_limit, record_order):
    return db.session.query(FuelConsumptionRecord).order_by(FuelConsumptionRecord.id.desc()).limit(records_limit)[record_order]


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return jsonify([FuelConsumptionRecord.serialize(record) for record in records])

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201


class FuelConsumption(Resource):
    def get(self, record_id):
        return jsonify(FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id))))

    def delete(self, record_id):
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        db.session.delete(record)
        db.session.commit()
        return '', 204

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201


class LastFuelConsumption(Resource):
    def get(self):
        last_record = get_record_by_order_desc(1, 0)
        second_last_record = get_record_by_order_desc(2, 1)
        distance = calculate_distance(second_last_record.odometer, last_record.odometer)
        consumption = round(calculate_consumption(last_record.fuelQuantity, distance), 2)
        return {'lastFuelConsumption': consumption}


class AverageFuelConsumption(Resource):
    def get(self):
        records_count = FuelConsumptionRecord.query.count()
        sum_of_consumptions = 0
        for i in reversed(range(0, records_count)):
            start_record = get_record_by_order_desc(records_count, i-1)
            end_record = get_record_by_order_desc(records_count, i-2)
            distance = calculate_distance(start_record.odometer, end_record.odometer)
            sum_of_consumptions += calculate_consumption(end_record.fuelQuantity, distance)
        avg_consumption = round(sum_of_consumptions / (records_count - 1), 2)
        return {"avgFuelConsumption": avg_consumption}


api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')
api.add_resource(LastFuelConsumption, '/calculateLastConsumption')
api.add_resource(AverageFuelConsumption, '/calculateAverageConsumption')

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

Code description

To calculate last fuel consumption and average foul consumption code from previous article is used.The only change is in retrieving data from database.

1. getrecordbyorderdesc is a helper method to retrieve records in descending order by id.

def get_record_by_order_desc(records_limit, record_order):
    return db.session.query(FuelConsumptionRecord).order_by(FuelConsumptionRecord.id.desc()).limit(records_limit)[record_order]

To retrieve data ordered by a field another approach is required. Unlike the previous implementation, there is no query directly on FuelConsumptionRecord class but methods are called on SQLAlchemy object.The cause of this change is possibility of use order_by method.

Testing

1. User is able to check last stored fuel consumption

Query:

curl -X GET http://localhost:5000/calculateLastConsumption 

Expected response:

    {
    "lastFuelConsumption": 8.57
    }

2. User is able to check average fuel consumption, based on all stored records

Query:

    curl -X GET http://localhost:5000/calculateAverageConsumption 

Expected response:

    {
    "avgFuelConsumption": 9.99
    }

Summary

Flask-RESTful with SQLAlchemy is a very good set for creating real world APIs. SQLAlchemy provides mechanisms for comprehensive query creation and to map Python objects to data in the database. In this article I showed you how easy is the process of implementing SQLAlchemy ORM to Flask-RESTful application.

Sources:

• https://flask-sqlalchemy.palletsprojects.com/en/2.x/
• https://flask-restful.readthedocs.io/