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.
Top comments (0)