DEV Community

ftarch
ftarch

Posted on • Edited on

4

Build Flask API using Marshmallow and SQLAlchemy

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
Enter fullscreen mode Exit fullscreen mode
  • Now activate the enviroment (I'm using linux, for windows activation see the virtualenv documentation)
.venv\Scripts\activate
Enter fullscreen mode Exit fullscreen mode
  • And now install the packages (ensure you have venv activated)
pip install flask flask-restful mariadb marshmallow flask-sqlalchemy
Enter fullscreen mode Exit fullscreen mode

Use this command to create the requirements.txt

pip freeze > requirements.txt
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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()




Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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 
Enter fullscreen mode Exit fullscreen mode

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)

Enter fullscreen mode Exit fullscreen mode

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

Enter fullscreen mode Exit fullscreen mode

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)

Enter fullscreen mode Exit fullscreen mode

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.

AWS Q Developer image

Your AI Code Assistant

Automate your code reviews. Catch bugs before your coworkers. Fix security issues in your code. Built to handle large projects, Amazon Q Developer works alongside you from idea to production code.

Get started free in your IDE

Top comments (0)

Billboard image

Create up to 10 Postgres Databases on Neon's free plan.

If you're starting a new project, Neon has got your databases covered. No credit cards. No trials. No getting in your way.

Try Neon for Free →

👋 Kindness is contagious

Please leave a ❤️ or a friendly comment on this post if you found it helpful!

Okay