How to Create a DRY RESTful API in Flask
Adnan Wazwaz
Introduction
If you have been learning how to build websites, then at some point, you may have come across terms like "API", "REST", and "backend". Learning how to build a backend is very important. But it's also important to do it in a standardized, consistent, and efficient manner. Here I will walk you through building a DRY Restful API using Flask.
Before diving into this, please ensure that you have a good understanding of Flask, Flask SQLAlchemy, and Flask Migrate. I also recommend to check out my article on server communication titled Server Communication: The Basics and its Example Application in JavaScript, which I have linked in the resources section at the end. Lastly, take time to understand the following terminology:
Backend
The "backend" is the stuff that an end user does not see. It's where an application's server exists in. It consists of an application to respond to clients' requests, and a database(s).
API
An application protocol interface (API), according to IBM's definition, is "a set of rules or protocols that enables software applications to communicate with each other to exchange data, features and functionality." Companies and organization can sometimes expose such data and functionality, so that programmers can use them in turn to power their own applications. Others will keep keep them private and block attempted fretch requests() on DevTool's console.
REST
Representational State Transfer (REST) is an architectural design pattern for standardized and consistent development of applications that use HTTP. An application that is built on REST principles is said to be "RESTful". There are a few benefits to building a RESTful API:
It eases client-server communication.
HTTP use is both human and machine-readable.
Network communication built on REST principles can be scaled.
For an API to be RESTful, it must satisfy the following five conditions:
Uniform Interface - for an API to have uniform interface, it must:
have consistent naming and exposition.
have only one logical URL.
be accessible using HTTP methods in a consistent format.
Stateless - a RESTful API does not keep track of client's requests.
Cacheable - A client stores information from a request, eliminating unneeded client-server interactions.
Client-Server - a RESTful application operates under the client-server model.
Layered System - where functionalities are assigned into layers that exist between clients and servers.
There is a sixth optional condition called code on demand, which is the ability for a server to send executable code to a client.
DRY
"Don't Repeat Yourself" (DRY) is the process of refactoring repeated code so that it becomes reusable.
Creating Resources
Let's say we have the following one-to-many relationship between students and report cards. Take some time to familiarize yourself with the schema and code below:
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy_serializer import SerializerMixin
from sqlalchemy.orm import validates
db = SQLAlchemy()
class Student(db.Model, SerializerMixin):
serialize_rules = ('-report_cards.student',)
__tablename__ = 'students'
id = db.Column(db.Integer, primary_key=True)
first_name = db.Column(db.String, nullable=False)
middle_name = db.Column(db.String)
last_name = db.Column(db.String, nullable=False)
email = db.Column(db.String, unique=True, nullable=False)
report_cards = db.relationship('ReportCard', back_populates='student', cascade='all, delete-orphan')
@validates('email')
def validate_email(self, key, email):
if "@" not in email:
raise ValueError(f"{key.title()} must be a real email address.")
elif Student.query.filter_by(email=email).first():
raise ValueError(f"Another student is already using {key}, {email}.")
return email
def __repr__(self):
return f"<Student {self.id}, {self.first_name}, {self.middle_name}, {self.last_name}, {self.email}>"
class ReportCard(db.Model, SerializerMixin):
__tablename__ = 'report_cards'
serialize_rules = ('-student.report_cards',)
id = db.Column(db.Integer, primary_key=True)
course_name = db.Column(db.String)
grade = db.Column(db.Integer, nullable=False)
student_id = db.Column(db.Integer, db.ForeignKey('students.id'))
student = db.relationship('Student', back_populates='report_cards')
__table_args__ = (
db.CheckConstraint('grade BETWEEN 0 and 100', name='check_grade_constraint'),
)
def __repr__(self):
return f"<ReportCard {self.id}, {self.course_name}, {self.grade}, {self.student_id}>"
Creating Views
When you started learning flask, you were probably exposed to creating resources using decorators ilke @app.route and defining a function for each resource. It would most likely look something like this:
@app.route('/students', methods=['GET', 'POST'])
def students():
if request.method == 'GET':
students = [student.to_dict() for student in Student.query.all()]
return make_response(students, 200)
elif request.method == 'POST':
new_student = Student(
first_name=request.form.get("first_name"),
middle_name=request.form.get("middle_name"),
last_name=request.form.get("last_name"),
email=request.form.get("email")
)
db.session.add(new_student)
db.session.commit()
return make_response(new_student.to_dict(), 201)
@app.route('/students/<int:id>', methods=['GET', 'PATCH', 'DELETE'])
def student_by_id(id):
student = Student.query.filter(Student.id == id).first()
if student == None:
response_body = {
"message": "This record does not exist in our database. Please try again."
}
return make_response(response_body, 404)
else:
if request.method == 'GET':
return make_response(student.to_dict(), 200)
elif request.method == 'PATCH':
for attr in request.form:
setattr(student, attr, request.form.get(attr))
db.session.add(student)
db.session.commit()
return make_response(student.to_dict(), 20-)
elif request.method == 'DELETE':
db.session.delete(student)
db.session.commit()
response_body = {
"delete_successful": True,
"message": "Student deleted."
}
return make_response(response_body, 204)
This gets tedious fast for two reasons. The first reason is that it invovles multiple conditional logic that slows down the application, and makes the code more difficult to follow. The second reason is that although, you needed to define only two functions and routes to handle five different CRUD methods for the Students model, you would have to repeat this process for the ReportCards model. Your code would be a tornado of boilerplate if you have many more models that are related to one another. This here is neither DRY nor efficient.
Using Flask RESful's Resource Class
We can solve the problem of speed and readability by refactoring our Flask views into Resources. A Resoure, in Flask, is an abstract container for a RESTful operation. You would have to create a new class that represents a concrete resource, and then have it inherit from Resource as a supercalss. The concrete resource is perfect for excluding it for a specific model in your database. Once you've defined a concrete resource, you will have access to CRUD methods, such as get, post, patch, and delete. One thing to do before running your application is to map the route to that class. Here's our updated API:
from flask import Flask, request, #make_response
from flask_migrate import Migrate
from flask_restful import Api, Resource
from models import *
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///school.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.json.compact = False
migrate = Migrate(app, db)
db.init_app(app)
api = Api(app)
class Students(Resource):
def get(self):
students = [student.to_dict() for student in Student.query.all()]
return students, 200
def post(self):
new_student = Student(
first_name=request.get_json().get("first_name"),
middle_name=request.get_json().get("middle_name"),
last_name=request.get_json().get("last_name"),
email=request.get_json().get("email")
)
db.session.add(new_student)
db.session.commit()
return new_student.to_dict(), 201
class StudentById(Resource):
def get(self, id):
student = Student.query.filter_by(id=id).first()
if student:
return student.to_dict(), 200
else:
return {'message': '404 Not Found'}, 404
def patch(self, id):
student = Student.query.filter_by(id=id).first()
if student is None:
return {'message': '404 Not Found'}, 404
for attr in request.get_json():
setattr(student, attr, request.get_json().get(attr))
db.session.add(student)
db.session.commit()
return make_response(student.to_dict(), 200)
def delete(self, id):
student = Student.query.filter_by(id=id).first()
if student is None:
return {'message': '404 Not Found'}, 404
db.session.delete(student)
db.session.commit()
return {}, 204
api.add_resource(Students, '/students', endpoint='students')
api.add_resource(StudentById, '/students/<int:id>', endpoint='student_by_id')
if __name__ == '__main__':
app.run(port=5555, debug=True)
This is much cleaner. Everything is organized into classes, where each class is a concrete resource and a sub-class of Resource. Those Each subclass is mapped to a specific route and endpoint. And there are much less conditional logic to worry about. Note the usage of plural Students instead of Student. I could have also declared it as StudentResource. Make sure to clearly differentiate between your resources and your models. Also note that we don't need to wrap our response with make_response(), when using resources.
Unfortunately, while this solves the first problem, it magnifies the second problem of repeated code. Firstly, in each method of StudentById, we are retrieving a student by id and returning an error message if not found. Secondly, we would have to still have to create Resources for ReportCard, and at that point, our code would no longer look clean, thus reproducing our first problem. So how can we solve this whole mess in its entirety?
DRY Implementation
The solution for an efficient, dry code requires a couple of steps.
Step #1: Isolating the Verbatim Repeated Code
The first step is to isolate code that we have repeated VERBATIM in each method of StudentById. In this context, I'm talking specifically about the retrieval of students by id and returning an error message if the student with the given id does not exist.
student = Student.query.filter_by(id=id).first()
if student is None:
return {'message': '404 Not Found'}, 404
It is clear that we need to retrieve the record before proceeding with any operation. So it would make sense to move this piece of code into a view with the @app.before_request decorator. It will be dedicated to getting any record of any model by id. We can get the correct model by mapping the endpoint to the model. To pass the record back to the actual CRUD method, we would have to import the g constant from flask and pass the record using g. Here's what it would look like:
from flask import Flask, request, g, make_response
from flask_migrate import Migrate
from flask_restful import Api, Resource
from models import *
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///school.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.json.compact = False
migrate = Migrate(app, db)
db.init_app(app)
api = Api(app)
@app.before_request
def get_record_by_id():
# Dictionary where keys are the endpoints, and the values are the models.
endpoint_model_map = {
'student_by_id': Student,
'report_card_by_id': ReportCard
}
if model := endpoint_model_map.get(request.endpoint):
id = request.view_args.get('id') # Retrieve argument values from a CRUD method.
if record := model.query.filter_by(id=id).first():
g.record = record
else:
return make_response({'message': '404 Not Found'}, 404)
# OTHER CODE
class StudentById(Resource):
def get(self, id):
student = g.record
return student.to_dict(), 200
def patch(self, id):
student = g.record
for attr in request.get_json():
setattr(student, attr, request.get_json().get(attr))
db.session.add(student)
db.session.commit()
return make_response(student.to_dict(), 200)
def delete(self, id):
student = g.record
db.session.delete(student)
db.session.commit()
return {}, 204
Now when you make a request to read or manipulate a student by a specific id, the get_record_by_id() view automatically runs first and intercepts the id. If there is a mathinc record, then the CRUD method will execute, otherwise, the error message is returned. At this point, we can go ahead and make resources for ReportCard.
class ReportCards(Resource):
def get(self):
report_cards = [report_card.to_dict() for report_card in Student.query.all()]
return report_cards, 200
def post(self):
new_rc = ReportCard(
course_name=request.get_json().get("course_name"),
grade=request.get_json().get("grade"),
student_id=request.get_json().get("student_id")
)
db.session.add(new_rc)
db.session.commit()
return new_rc.to_dict(), 201
class ReportCardById(Resource):
def get(self, id):
rc = g.record
return rc.to_dict(), 200
def patch(self, id):
rc = g.record
for attr in request.get_json():
setattr(rc, attr, request.get_json().get(attr))
db.session.add(rc)
db.session.commit()
return make_response(rc.to_dict(), 200)
def delete(self, id):
rc = g.record
db.session.delete(rc)
db.session.commit()
return {}, 204
api.add_resource(ReportCards, '/report_cards', endpoint='report_cards')
api.add_resource(ReportCardById, '/report_cards/<int:id>', endpoint='report_card_by_id')
Step #2: Isolating Similarly Repeated Code
We got rid of as much verbatim repeated code as possible. But the challenge now is to isolate repeated code that is not verbatim, but whose statements are close to it. In our example, it would just be the rest of our code in each crud method. The best way to isolate this is to create a "template" class that inherits from Resource, where we can pass a model class to the constructor. From there, we can re-write almost all our CRUD methods in that template without ever having to specify the name of the model itself.
class DRYResource(Resource):
def __init__(self, model):
self.model = model
def get(self, id = None):
if type(id) is int:
record = g.record
return record.to_dict(), 200
else:
records = [record.to_dict() for record in self.model.query.all()]
return records, 200
def patch(self, id):
try:
record = g.record
json = request.get_json()
for attr in json:
#print(f"Attrname: {attr}, Type: {type(json.get(attr))}")
setattr(record, attr, json.get(attr))
db.session.add(record)
db.session.commit()
return record.to_dict(), 200
except ValueError as e:
print(e)
return {'error': 'Not Modified'}, 304
def delete(self, id):
record = g.record
db.session.delete(record)
db.session.commit()
return {'message': f'{self.model.__name__} successfully deleted.'}, 204
The before_request view, get_record_by_id() stil executes before any of these CRUD methods. POST is not rewritten under DRYResource because each model has their own different set of attributes, values, checks, and validations that require to be declared in their own modal-mapped resource. You can attempt to create a DRY post method in the template, and I would encourage you to do so. Feel free to let me know what you come up with in the comments. However, to keep our example simple, we will disregard that.
One other thing to note is that this does not take into account of applying serialization rules when converting objects to JSONified dictionaries. Again, you can experiment with that on your own, but I will not cover that here.
Step #3: Refactor Concrete Resources
Now that we have our template, we need to have our concrete resource inherit from DRYResource instead of Resource. Then, we need to pass the appropriate models into the superclass. Finally, we can remove all GET, PATCH, and DELETE methods since we would be retrieving our data from the superclass's methods. We need to keep POST due to reasons discussed in the last step.
class Students(DRYResource):
def __init__(self):
super().__init__(Student)
def post(self):
new_student = Student(
first_name=request.get_json().get("first_name"),
middle_name=request.get_json().get("middle_name"),
last_name=request.get_json().get("last_name"),
email=request.get_json().get("email")
)
db.session.add(new_student)
db.session.commit()
return new_student.to_dict(), 201
class StudentById(DRYResource):
def __init__(self):
super().__init__(Student)
class ReportCards(DRYResource):
def __init__(self):
super().__init__(ReportCard)
def post(self):
new_rc = ReportCard(
course_name=request.get_json().get("course_name"),
grade=request.get_json().get("grade"),
student_id=request.get_json().get("student_id")
)
db.session.add(new_rc)
db.session.commit()
return new_rc.to_dict(), 201
class ReportCardById(DRYResource):
def __init__(self):
super().__init__(ReportCard)
Complete Final Version
This is what the final version looks like in its entirety. Go ahead and experiment this with Postman to show that it really works.
from flask import Flask, request, g, make_response
from flask_migrate import Migrate
from flask_restful import Api, Resource
from models import *
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///school.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.json.compact = False
migrate = Migrate(app, db)
db.init_app(app)
api = Api(app)
@app.before_request
def get_record_by_id():
# Dictionary where keys are the endpoints, and the values are the models.
endpoint_model_map = {
'student_by_id': Student,
'report_card_by_id': ReportCard
}
if model := endpoint_model_map.get(request.endpoint):
id = request.view_args.get('id') # Retrieve argument values from a CRUD method.
if record := model.query.filter_by(id=id).first():
g.record = record
else:
return make_response({'message': '404 Not Found'}, 404)
class DRYResource(Resource):
"""
Template for RESTful CRUD methods.
"""
def __init__(self, model):
"""Creates a new instance of RestResourceTemplate.
Args:
model (db.Model): the model to tie the resource to.
"""
self.model = model
def get(self, id = None):
"""
If an id is specified, then retrieves a db.Model record with that id if it exists.
If the record with that id does not exist, then returns a message saying "not found".
Otherwise, returns all records.
"""
if type(id) is int:
record = g.record
return record.to_dict(), 200
else:
records = [record.to_dict() for record in self.model.query.all()]
return records, 200
def patch(self, id):
"""Updates a db.Model record with a given id.
Args:
id (int): the id.
Returns:
dict: a JSONified dictionary of the record if successfully updated; if update failed, then will
return a "Not Modified" message. If record with id, does not exist, a "Not Found" will be returned.
"""
try:
record = g.record
json = request.get_json()
for attr in json:
#print(f"Attrname: {attr}, Type: {type(json.get(attr))}")
setattr(record, attr, json.get(attr))
db.session.add(record)
db.session.commit()
return record.to_dict(), 200
except ValueError as e:
print(e)
return {'error': 'Not Modified'}, 304
def delete(self, id):
"""Deletes a db.Model record with a given id.
Args:
id (int): the id.
Returns:
dict: no content or a message saying that the record was deleted.
"""
record = g.record
db.session.delete(record)
db.session.commit()
return {'message': f'{self.model.__name__} successfully deleted.'}, 204
class Students(DRYResource):
def __init__(self):
super().__init__(Student)
def post(self):
new_student = Student(
first_name=request.get_json().get("first_name"),
middle_name=request.get_json().get("middle_name"),
last_name=request.get_json().get("last_name"),
email=request.get_json().get("email")
)
db.session.add(new_student)
db.session.commit()
return new_student.to_dict(), 201
class StudentById(DRYResource):
def __init__(self):
super().__init__(Student)
class ReportCards(DRYResource):
def __init__(self):
super().__init__(ReportCard)
def post(self):
new_rc = ReportCard(
course_name=request.get_json().get("course_name"),
grade=request.get_json().get("grade"),
student_id=request.get_json().get("student_id")
)
db.session.add(new_rc)
db.session.commit()
return new_rc.to_dict(), 201
class ReportCardById(DRYResource):
def __init__(self):
super().__init__(ReportCard)
api.add_resource(Students, '/students', endpoint='students')
api.add_resource(StudentById, '/students/<int:id>', endpoint='student_by_id')
api.add_resource(ReportCards, '/report_cards', endpoint='report_cards')
api.add_resource(ReportCardById, '/report_cards/<int:id>', endpoint='report_card_by_id')
if __name__ == '__main__':
app.run(port=5555, debug=True)
Conclusion
And that's all there is for creating an efficient, DRY, reusable template for RESTful API's. You can work from this blueprint to implement any advanced DRY designs such as handling complex POST validation and serialization rules. Check out the sources I linked below if you would like to to learn more.
Sources
Server Communication: The Basics and its Example Application in JavaScript
Front-End vs. Back-End: What’s the Difference?
Subscribe to my newsletter
Read articles from Adnan Wazwaz directly inside your inbox. Subscribe to the newsletter, and don't miss out.
Written by