REST is the most common way to structure web APIs over HTTP these days. It is based on modeling everything as resources identified by URLs and manipulating the resources using the HTTP verbs GET, POST, PUT and DELETE. Flask is an extensible Python micro-framework for web development. You can develop a REST API using Flask on its own, but, the Flask-RestFUL extension directly supports REST API development by exposing a resource-based approach. In this article I’ll explore the various components of a REST API built on top of Flask-RestFUL via a simple example.
Our example projects is called “Over Achiever” and allows you to get, create, update and delete goals and sub-goals that can be nested arbitrarily.
Each user can have his or her own elaborate tree-like structure of nested goals and subgoals. A goal can be marked as complete by providing an end timestamp. The full source code can be found here.
Models
I model the domain as SQLAlchemy models. This has the nice property of mapping resources directly to rows in a database. It is a very common pattern and makes many tasks trivial. GET, POST, PUT and DELETE REST operations on resource X often are simply SELECT, INSERT, UPDATE and DELETE operation on a single table. Note, that often you would want to operate on memory models or process data before/after storing/retrieving.
Let’s start. There are two concepts we want to model a “user” and a goal”. We’ll put them in a file called models.py
from datetime import datetimefrom sqlalchemy import (Column, DateTime, ForeignKey, Integer, String)from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import relationshipBase = declarative_base()metadata = Base.metadataclass User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True) name = Column(String(64), unique=True) email = Column(String(120), index=True, unique=True) password = Column(String(128))class Goal(Base): __tablename__ = 'goal' id = Column(Integer, primary_key=True) user_id = Column(ForeignKey('user.id'), nullable=False) parent_id = Column(ForeignKey('goal.id')) name = Column(String(64), unique=True) description = Column(String(512)) start = Column(DateTime, default=datetime.now) end = Column(DateTime, nullable=True) user = relationship('User') parent = relationship(lambda: Goal, remote_side=id, backref='sub_goals')The User model is completely trivial and stores user name, role and credentials. The Goal model is more interesting. It stores goal name, description, start and end time and the reference to the user that defined this goal. But, it also has a parent reference to the same goal table. This self-reference of sorts is how the hierarchical nature of goals is accomplished in the relational database. While, in programming language you would typically model it as each goal having a list of sub-goals here it is modeled differently where each goal may have a parent goal. A goal with no parent is a top-level goal.
Resources
Resources are a Flask-Restful concept. Each resource may have a get(), post(), put() and delete() method. Each resource directly support a specific REST endpoint and the methods implement the HTTP verbs for this resource. All the methods are optional.
Here is the Goal Resource class:
class Goal(Resource): def get(self): """Get all goals organized by user and in hierarchy""" q = db.session.query result = {} users = q(m.User).all() for u in users: result[u.name] = _get_goal_tree(q, u) return result def post(self): parser = RequestParser() parser.add_argument('user', type=str, required=True) parser.add_argument('name', type=str, required=True) parser.add_argument('parent_name', type=str) parser.add_argument('description', type=str, required=False) args = parser.parse_args() # Get a SQL Alchemy query object q = db.session.query # Create a new goal user = q(m.User).filter_by(name=args.user).one() # Find parent goal by name if args.parent_name: parent = q(m.Goal).filter_by(name=args.parent_name).one() else: parent = None goal = m.Goal(user=user, parent=parent, name=args.name, description=args.description) db.session.add(goal) db.session.commit() def put(self): """Update end time""" parser = RequestParser() parser.add_argument('name', type=str, required=True) args = parser.parse_args() # Get a SQL Alchemy query object q = db.session.query goal = q(m.Goal).filter_by(name=args.name).one() goal.end = datetime.now() db.session.commit()The RequestParser allows you to specify and parse the request parameters and then use them in the implementation of each method. The post(), put() and delete() methods don’t return anything, which if successful will result in a simple 200 HTTP status code on the caller side. If anything goes wrong you can return any HTTP error code via Flask’s abort() method.
The API
The api.py module contains the entry point to the service, initializes several critical components like the configuration and the database and maps routes (HTTP end points) to resources:
import osfrom flask import Flaskfrom flask_restful import Apifrom flask.ext.sqlalchemy import SQLAlchemyfrom resources import User, Goalimport resourcesdef create_app(): app = Flask(__name__) app.config.from_object('over_achiever.config') resources.db = app.db = SQLAlchemy(app) api = Api(app) resource_map = ( (User, '/v1.0/users'), (Goal, '/v1.0/goals'), ) for resource, route in resource_map: api.add_resource(resource, route) return appthe_app = create_app()if __name__ == "__main__": print("If you run locally, browse to localhost:5000") host = '0.0.0.0' port = int(os.environ.get("PORT", 5000)) the_app.run(host=host, port=port) The create_app() function is the key. It mixes together all the ingredients and return an application objects that is ready to run. Then in the __main__ section the run() method is called, listening on all interfaces (ip address 0.0.0.0) and using port 5000 or the PORT environment variable (needed for Heroku).
Testing
The tests serve two functions:
- Make sure the code actually works as intended
- Show how to use the API in practice
I use the standard unittest module for implementing the test and the provided Flask test client (via the app.test_client()) to invoke the API through its end points:
class OverAchieverTest(TestCase): def setUp(self): self.app = create_app() self.session = create_mem_db(metadata, self.app.db) self.test_app = self.app.test_client() # add users u1 = m.User(name='user-1', email='[email protected]', password='123') u2 = m.User(name='user-2', email='[email protected]', password='123') self.user = m.User(name='user-3', email='[email protected]', password='123') self.session.add(u1) self.session.add(u2) self.session.add(self.user) # add goals goals = [None] * 8 goals[0] = m.Goal(user=u1, name='goal-0') goals[1] = m.Goal(user=u1, name='goal-1') goals[2] = m.Goal(user=u1, name='goal-2', parent=goals[1]) goals[3] = m.Goal(user=u1, name='goal-3', parent=goals[1]) goals[4] = m.Goal(user=u1, name='goal-4', parent=goals[3]) goals[5] = m.Goal(user=u1, name='goal-5', parent=goals[3]) goals[6] = m.Goal(user=u1, name='goal-6', parent=goals[3]) goals[7] = m.Goal(user=u2, name='goal-7') for g in goals: self.session.add(g) self.session.commit() def tearDown(self): pass def test_get_goals(self): #q = self.session.query url = '/v1.0/goals' response = self.test_app.get(url) result = json.loads(response.data) expected = {'user-1': {'goal-0': {}, 'goal-1': {'goal-2': {}, 'goal-3': { 'goal-4': {}, 'goal-5': {}, 'goal-6': {}}}}, 'user-2': {'goal-7': {}}, 'user-3': {}} self.assertEqual(expected, result) def test_add_new_goal(self): q = self.session.query name = 'new-goal' # verify the goal doesn't exist yet self.assertIsNone(q(m.Goal).filter_by(name=name).scalar()) params = dict(user=self.user.name, name=name) url = '/v1.0/goals' response = self.test_app.post(url, data=params) self.assertEqual(200, response.status_code) self.assertIsNotNone(q(m.Goal).filter_by(name=name).scalar()) def test_add_nested_goals(self): q = self.session.query name = 'new-goal' # verify the goal doesn't exist yet self.assertIsNone(q(m.Goal).filter_by(name=name).scalar()) params = dict(user=self.user.name, name=name) url = '/v1.0/goals' response = self.test_app.post(url, data=params) self.assertEqual(200, response.status_code) parent_goal = q(m.Goal).filter_by(name=name).one() child_name = 'child' params = dict(user=self.user.name, name=child_name, parent_name=parent_goal.name) response = self.test_app.post(url, data=params) self.assertEqual(200, response.status_code) child_goal = q(m.Goal).filter_by(name=child_name).one() self.assertEqual(child_name, child_goal.name) self.assertEqual(parent_goal, child_goal.parent) def test_complete_goal(self): q = self.session.query params = dict(user='user-1', name='goal-0') url = '/v1.0/goals' response = self.test_app.put(url, data=params) self.assertEqual(200, response.status_code) goal = q(m.Goal).filter_by(name='goal-0').one() self.assertIsNotNone(goal.end) Conclusion
This is a very minimal example of a REST API, but it contains all the pieces and you can use it as a starting point for more ambitious projects. If you feel comfortable with Python, Flask and SQLAlchemy then more power to you.
