devxlogo

Writing RESTful Web Services in Python with Flask

Writing RESTful Web Services in Python with Flask

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:

  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='[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.

devxblackblue

About Our Editorial Process

At DevX, we’re dedicated to tech entrepreneurship. Our team closely follows industry shifts, new products, AI breakthroughs, technology trends, and funding announcements. Articles undergo thorough editing to ensure accuracy and clarity, reflecting DevX’s style and supporting entrepreneurs in the tech sphere.

See our full editorial policy.

About Our Journalist