CRUD Application with FLASK
So in my last blog, I told you how to write SQL based applications.here I will tell you the RESTful Flask application from scratch. Here I will maintain a database of
Author objects and it will also have the user authentication mechanism to only allow logged-in users to use certain functions.
I will create the following endpoints for our REST applications:
1) GET/authors: get lists of books
2)GET/authors/<id>: This gets author with the certain ID .
3)POST/authors: This creates a new author object.
4)PUT/authors/<id>: This will edit autor object with the given ID.
5)DELETE/authors/<id>: This will delete the author with given ID.
6)GET/books : this will return all the books.
7)GET/books/<id>: this gets the book with specified ID.
8)POST/books: This creates a new book object.
9)PUT/books/<id>: This will edit book object with specified ID.
10)DELETE/book/<id>: This will delete the book with given ID.
Now let's create it. Create a new project and name it author-manager.
So create new virtual environment.
$ mkdir author-manager && cd author-manger
$ virtualenv venv
Now install the following dependencies.
(venv) $ pip install flask flask-sqlalchemy marshmallow-sqlalchemy
Just an info FLASK uses concept of blueprints to make application components and support common patterns across the application. The blueprint helps create smaller modules for
larger application and simplifies how large application works.
So we will structure the application small modules and keep all our application code in the /src folder inside our app folder.
Now we will create an src folder inside our current directory and then create run.py files inside it.
(venv) $ mkdir src && cd src
We will initiate our FLASK app in main.py file inside the src and then create another file run.py which will import main.py, config file, and run the application.
main.py :
import os
from flask import Flask
from flask import jsonify
app = Flask(_name_)
if os.environ.get('WORK_ENV') == 'PROD'
app_config = ProductionConfig
elif os.environ.get('WORK_ENV') == 'TEST'
app_config = TestingConfig
else:
app_config = DevelopmentConfig
app.config.from_object(app_config)
if _name_ == "_main_":
app.run(port=5000,host="0.0.0.0",use_reloader=False)
This is how our main.py will look like and next we will create our run.py to call app and run the application.
run.py:
from main import app as application
if _name_ == "_main_":
application.run()
Here we defined the config, improve create_app, and initialized the application. Now we will create another directory/API inside src and export config, models, and routes from API directory.
so now create a directory inside src called API and then create another directory called config inside API.
config.py:
class Config(object):
DEBUG = False
TESTING = False
SQLALCHEMY_TRACK_MODIFICATIONS = False
class ProductionConfig(Config):
SQLALCHEMY_DATABASE_URI = <Production DB URL>
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = <Development DB URL>
SQLALCHEMY_ECHO = False
class TestingConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = <Testing DB URL>
SQLALCHEMY_ECHO = False
Now create a database.py:
from flask-sqlalchemy import SQLAlchemy
db = SQLAlchemy()
And this shall initiate to create our db object.now will import the main.py and initialize it.
from api.database import db
def create_app(config):
app = Flask(_name_)
app.config.from_object(config)
db.init_app(app)
with app.app_context():
db.create_all()
return app
And update create_app to initialize the db object.
So till now, we have an structure like below:
venv/
src
api/
_init_.py
database.py
config/
_init_.py
database.py
run.py
main.py
requirements.txt
Now let's define our db schema. So we will deal with two resources author and book. Let's create book models.
books.py:
from api.database import db
from marshmallow-sqlalchemy import ModelSchema
from marshmallow import fields
class Book(db.Model):
_tablename_ = 'books'
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
title = db.Column(db.String(50))
year= db.Column(db.Integer)
author_id = db.Column(db.Integer,db.ForeignKey('authors.id'))
def _init_(self,title,year,author_id=None):
self.title= title
self.year= year
self.author_id = author_id
def create(self):
db.session.add(self)
db.session.commit()
return self
class BookSchema(ModelSchema) :
class Meta(ModelSchema.Meta) :
model = Book
sqla_session = db.session
id = feilds.Number(dump-only=True)
title = feilds.String(required=True)
year= feilds.integer(required=True)
author_id = feilds.Integer()
Here we are importing db module marshmallow-like we did in our last blog and return JSON objects.
authors.py:
from api.database import db
from marshmallow-sqlalchemy import ModelSchema
from marshmallow import fields
from api.models.books import BookSchema
class Author(db.Model):
_tablename_ = 'authors'
id = db.Column(db.Integer,primary_key=True,autoincrement=True)
first_name = db.Column(db.String(50))
last_name= db.Column(db.String(50))
created = db.Column(db.DateTime,server_default=db.func.noe())
books = db.relationship('Books',backref='Author',cascade="all, delete-orphan")
def _init_(self,first_name,last_name,books=[]):
self.first_name= first_name
self.last_name= last_name
self.books = books
def create(self):
db.session.add(self)
db.session.commit()
return self
class AuthorSchema(ModelSchema):
class Meta(ModelSchema.Meta) :
model = Author
sqla_session = db.session
id = feilds.Number(dump_only=True)
first_name = feilds.String(required=True)
last_name= feilds.String(required=True)
created = feilds.String(dump_only=True)
books = feilds.Nested(BookSchema,many=True,only=['title','year','id'])
So after all this, we have to write global HTTP configurations in main.py
Now create responses.py inside the api and use jsonify and make_response from Flask library to create standard responses for our APIs.
responses.py:
from flask import make_response,jsonify
def resposnes_with(response,value=None,message=None,error=None,headers={},pagination=None):
result= {}
if value is not None:
result.update(value)
if response.get('message',None) is not None:
result.update({'message:' response['message']})
result.update({'code': response['code'] })
if error is not None:
result.update({'errors':error})
if pagination is not None:
result.update('pagination':pagination)
headers.update({'Access-Control-Allow-Orgin': '*'})
headers.update({'server': 'Flask REST API'})
return make_response(jsonify(result),response['http_code'],headers)
This exposes a function reponse_with for our API endpoints to use and respond back.
HTTP responses :
200 200 Ok - Standard response to HTTP requests
201 201 Created - A new resource has been created
204 204 No Content - No data returned a successful request.
400 400 Bad Request - Server cant process the error due to client error
403 403 Not Authorized - not authorized to obtain the resources
404 404-Not Found - Resource does not exist on the server
422 422 Unprocessable Entity - Due to semantic error
500 500 Internal Server Error - Unexpected condition in server.
INVALID_FEILD_NAME_SENT_422 = {
"http_code" : 422,
"code":"InvalidDeild",
"message":"Invalid Feilds Found"
}
INVALID_INPUT_422 = {
"http_code" : 422,
"code":"InvalidInput",
"message":"Invalid Input"
}
MISSING_PARAMETERS_422 = {
"http_code" : 422,
"code":"missingParameters",
"message":"Missing parameters"
}
BAD_REQUEST_400 = {
"http_code" : 400,
"code":"badRequest",
"message":"Bad request"
}
SERVER_ERROR_500 = {
"http_code" : 500,
"code":"notFound",
"message":"Server Error!"
}
SERVER_ERROR_404 = {
"http_code" : 404,
"code":"notFound",
"message":"Resource not found!"
}
UNAUTHORIZED_403 = {
"http_code" : 403,
"code":"notAuthorized",
"message":"you are not authorized to execute this!"
}
SUCCESS_200 = {
"http_code" : 200,
"code":"success",
}
SUCCESS_201 = {
"http_code" : 201,
"code":"success",
}
SUCCESS_204 = {
"http_code" : 204,
"code":"success",
}
Now we have our working respose.py model ready but we have to add some to our main.py
from api.responses import response_with
import api.responses as resp
just add above db.init_app function.
@app.after_request
def add_header(response):
return response
@app.errorhandler(400)
def bad_request(e):
logging.error(e)
return response_with(resp.BAD_REQUEST_400)
@app.errorhandler(500)
def server_error(e):
logging.error(e)
return response_with(resp.SERVER_ERROR_500)
@app.errorhandler(404)
def server_error(e):
logging.error(e)
return response_with(resp.SERVER_ERROR_404)
Now we need to create our api endpoints and include them in our main.py
from flask import Blueprint
from flask import request
from api.responses import response_with
from api import responses as resp
from api.models.authors import Author ,AuthorSchema
from api.databases import db
author_routes = Blueprint("author_routes",_name_)
@author_routes.route('/',methods=['POST'])
def create_author():
try:
data = reuest.get_json()
author_schema= AuthorSchema
author,error=author_schema.load(data)
result=author_schema.dump(author.create()).data
return response_with(resp.SUCCESS_201,value={"author": result})
except Exception as e:
print e
return response_with(resp.INVALID_INPUT_422)
So it will take JSON data from the request and execute create method on the Author schema and then return the response using response_with method.
So in our main.py import the author routes and the registers the blueprint.
from api.routes.authors import author_routes
app.register_blueprint(author_routes,url_prefix='/api/authors')
Now run the application using Python command run.py and our flask server should be up and running.
Now let's try with POSTMAN with the below URL.
https://localhost:5000/api/authors/ with following JSON data.
{
"first_name": "Abhinav",
"last_name": "Tripathi"
}
So first is ready to use check it.
