Build Flask API using Marshmallow and SQLAlchemy
Tags: marshmallow, flask, mariadb, sqlalchemy
Published: 2023-01-17 21:57:08

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:

```python
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.

```python
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

```python
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:

```python
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:

```python
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:

```python
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. 

```python
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.