Login | Register   
LinkedIn
Google+
Twitter
RSS Feed
Download our iPhone app
TODAY'S HEADLINES  |   ARTICLE ARCHIVE  |   FORUMS  |   TIP BANK
Browse DevX
Sign up for e-mail newsletters from DevX


advertisement
 

Writing RESTful Web Services in Python with Flask

Explore the various components of a REST API built on top of Flask-RestFUL via a simple example.


advertisement

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 datetime
from sqlalchemy import (Column,
                        DateTime,
                        ForeignKey,
                        Integer,
                        String)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()
metadata = Base.metadata


class 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 os
from flask import Flask
from flask_restful import Api
from flask.ext.sqlalchemy import SQLAlchemy
from resources import User, Goal
import resources

def 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 app
the_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:

  1. Make sure the code actually works as intended
  2. 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='user-1@example.org',
                    password='123')
        u2 = m.User(name='user-2',
                    email='user-2@example.org',
                    password='123')

        self.user = m.User(name='user-3',
                           email='user-3@example.org',
                           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.



   
Gigi Sayfan is the chief platform architect of VRVIU, a start-up developing cutting-edge hardware + software technology in the virtual reality space. Gigi has been developing software professionally for 21 years in domains as diverse as instant messaging, morphing, chip fabrication process control, embedded multi-media application for game consoles, brain-inspired machine learning, custom browser development, web services for 3D distributed game platform, IoT/sensors and most recently virtual reality. He has written production code every day in many programming languages such as C, C++, C#, Python, Java, Delphi, Javascript and even Cobol and PowerBuilder for operating systems such as Windows (3.11 through 7), Linux, Mac OSX, Lynx (embedded) and Sony Playstation. His technical expertise includes databases, low-level networking, distributed systems, unorthodox user interfaces and general software development life cycle.
Comment and Contribute

 

 

 

 

 


(Maximum characters: 1200). You have 1200 characters left.

 

 

Sitemap
Thanks for your registration, follow us on our social networks to keep up-to-date