Build Flask API using Marshmallow and SQLAlchemy

ftarch — Tue, 17 Jan 2023 21:57:08 +0000

Hello there!

In this article we will build a simple restful-API using flask, sqlalchemy and marshmallow.

This is the project folders structure.

├── App
│ ├── models
│ │ ├── init.py
│ │ ├── car_model.py
│ ├── resources
│ │ ├── init.py
│ │ ├── car_resource.py
│ ├── schemas
│ │ ├── init.py
│ │ ├── car_schema.py
├── app.py
├── config.py
├── routes.py
├── requirements.txt

Follow the steps bellow and in the end you have an endpoint running the application in your http://localhost:5000/

Virtualenv and packages installation

I will use virtualenv to create an enviroment to this project.

virtualenv .venv

Now activate the enviroment (I'm using linux, for windows activation see the virtualenv documentation)

.venv\Scripts\activate

And now install the packages (ensure you have venv activated)

pip install flask flask-restful mariadb marshmallow flask-sqlalchemy

Use this command to create the requirements.txt

pip freeze > requirements.txt

Setup Flask

In the app.py we start the flask application:

from flask import Flask


def app_start():

    app = Flask(__name__)
    return app


APP = app_start()

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

Integrate Flask-SQLAlchemy and route resources

I'm using SQLAlchemy to connect to mariadb session, you can use your favorite database here. I use mariadb because already have it installed in my system. You can use sqlite3 to simplify.

Open the config.py and then configure the SQLAlchemy connection.

from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()


def config_database(app):

    app.config['SQLALCHEMY_DATABASE_URI'] = \
    '{DBMS}://{user}:{password}@{server}/{database}'.format(
        DBMS='mariadb+mariadbconnector',
        user='root',
        password='123456',
        server='localhost',
        database='cars'
    )
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

    db.init_app(app)
    with app.app_context():
        db.create_all()

Now config the routes in routes.py

from flask_restful import Api
from resources.car_resource import CrudRoute


def config_routes(app):
    api = Api()

    api.add_resource(CrudRoute, '/', '/<int:identifier>', methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'])

    api.init_app(app)

Create models

Now open the car_model.py from models folder and code the database model:

from config import db


class Car(db.Model):
    __tablename__ = "Cars"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    brand = db.Column(db.String(255))
    name = db.Column(db.String(255))
    year = db.Column(db.Integer)
    color= db.Column(db.String(255))
    type = db.Column(db.String(255))

    def __init__(self, brand, name, year, color, type):
        self.brand = brand 
        self.name = name 
        self.year = year 
        self.color = color
        self.type = type

Integrate Flask-Marshmallow

In the schemas folder open the car_schema.py and use marshmallow to validate json data:

from marshmallow import Schema, fields


class CarSchema(Schema):
    identifier = fields.Int()
    brand = fields.Str(required=True)
    name = fields.Str(required=True)
    year = fields.Int(required=True)
    color = fields.Str(required=True)
    type = fields.Str(required=True)


car_schema = CarSchema()
cars_schema = CarSchema(many=True)

Create the CRUD

We will build a simple crud to interact with the application.
Inside the resources folder open the car_resouce.py and code this:

from flask_restful import Resource
from flask import request
from schemas.car_schema import cars_schema, car_schema
from config import db
from models.car_model import Car


class CrudRoute(Resource):
    def get(self, identifier=None):
        if identifier is not None:
            car = Car.query.get(identifier)
            return car_schema.dump(car)
        else:
            cars = Car.query.all()
            return cars_schema.dump(cars)

    def post(self):
        json_data = request.get_json()
        if not json_data:
            return {'message': 'Error: expected data not received!'}, 400

        data = car_schema.load(json_data)
        car = Car(**data)

        db.session.add(car)
        db.session.commit()
        result = car_schema.dump(car)

        return {"status": 'Success', 'data': result}, 201

    def put(self, identifier):
        json_data = request.get_json()
        if not json_data:
            return {'message': 'Error: expected data not received'}, 400

        data = car_schema.load(json_data)
        car = Car.query.get(identifier)
        car.brand= data['brand']
        car.name = data['name']
        car.type= data['type']
        car.year = data['year']
        car.color = data['color']

        db.session.commit()
        result = car_schema.dump(carro)
        return {"status": 'Success', 'data': result}, 204

    def patch(self, identifier):
        car = Car.query.get(identifier)
        if not car:
            return {'message': 'Car not found!'}, 404

        json_data = request.get_json()
        if not json_data:
            return {'message': 'Error: expected data not received'}, 400

        data = carro_schema.load(json_data, partial=True)
        for key, value in data.items():
            setattr(car, key, value)

        db.session.commit()
        result = car_schema.dump(car)

        return {"status": 'Success', 'data': result}, 204

    def delete(self, identifier):
        car = Car.query.get(identifier)
        db.session.delete(car)
        db.session.commit()

        return {'message': 'Car deleted!'}, 204

Run the application

In this project i'm using a simple factory pattern, a function to call the configs.
Get back to app.py and call the routes and database config to start the application.

from flask import Flask
from resources.car_resource import CrudRoute
from config import config_database
from routes import config_routes


def app_start():

    app = Flask(__name__)

    config_routes(app)
    config_database(app)

    return app


APP = app_start()

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

Type python app.py to run the application in the http://localhost:5000.

We are running the api, now you can test it with Insomnia or Postman.

DEV Community: ftarch

Build Flask API using Marshmallow and SQLAlchemy