October 4, 2014

Writing a RESTful API in Flask & SQLAlchemy

First Thing's First

Why would we write this in Python? There's already rails, spring, and the mean stack which can do the same thing. Let's consider what Python gives us out of the box. Out of all the languages I've touched, it definitely has the easiest learning curve. The syntax is beautiful, and the "batteries included" principle of python just helps you get your server up very quick. Install python, find your dependencies, get coding.

Heads up! This article is deprecated. It may not apply to the latest versions of Flask/SQLAlchemy.

Getting Started

For this tutorial, we're going to use Python 2.7 and Flask. We're also going to use a sqlite database and SQLAlchemy. Let's get started... For simplicity's sake, we're just going to keep our code in two files.

Config.py
    import os
    basedir = os.path.abspath(os.path.dirname(__file__))

    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'app.sqlite')
    SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository')

Config.py just sets up some basic necessities so we can connect to our database. Although this isn't really doing much right away, we're going to load this configuration into our app in just a moment.

Flask.py
    from flask import Flask, jsonify, abort, request
    from flask.ext.sqlalchemy import SQLAlchemy

    app = Flask(__name__)
    app.config.from_pyfile('config.py')
    db = SQLAlchemy(app)    

This covers the basics of our app. We're able to import our necessities (Flask, SQLAlchemy, and OS) and then instantiate our Flask application with Flask(__name__). Our configuration from earlier is loaded into our application with app.config.from_pyfile('config.py') and lastly, we create our database by instantiating a SQLAlchemy instance with our app as the one parameter.

Models

Staying in the same file (Flask.py), we're going to create a model for our database. Our model will look like this.

    class Developer(db.Model):
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(20))
        hireDate = db.Column(db.Date)
        focus = db.Column(db.String(50))

class Developer(db.Model): creates a new class named Developer (this ties directly to a table in our database called "Developer") and gives us an object to create/update/delete in our api. The code within the class just explains the properties on the class and how they map to the database, for example, id maps to a column in our database that is an integer and is also our primary_key while name maps to just a string field.

Flask.py (Checkpoint)

Up to this point, our Flask.py should look like the following:

    from flask import Flask, render_template
    from flask.ext.restless import APIManager
    from flask.ext.sqlalchemy import SQLAlchemy

    app = Flask(__name__)
    app.config.from_pyfile('config.py')
    db = SQLAlchemy(app)

    class Developer(db.Model):
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(20))
        hireDate = db.Column(db.Date)
        focus = db.Column(db.String(50))

        def __init__(self, name, hireDate, focus):
            self.name = name
            self.hireDate = datetime.datetime.strptime(hireDate, "%d%m%Y").date()
            self.focus = focus

We're going to add one more line to that file before we start handling requests to the server. Simply add the following: db.createAll() . What db.createAll() does is simple, it runs against our models and says, if we need to create tables for these models (in this case, just developer), create the tables. With all of this code, we're ready to start handling routes.

    @app.route('/dev/', methods = ['GET'])
    def index():
        return jsonify({'developers': Developer.query.all()})

The above route simple says, if someone hits http://ourserver/dev/ they'll get back all of the developers in our database. jsonify simply converts the object into a pretty snazzy json object.

    @app.route('/dev/<int:id>/')
    def get_dev(id):
        return jsonify({'developer': Developer.query.get(id)})

This route runs very similar to the main index route, but with a caveat. This one takes an id in the url, and only returns a json object with the developer whos primary key matches the one passed in.

    @app.route('/dev/', methods = ['POST'])
    def create_dev():
        if not request.json or not 'name' in request.json:
            abort(400)
        dev = Developer(request.json.name, request.json.get('hireDate', ''), request.json.get('focus',''))
        db.session.add(dev)
        db.session.commit()
        return jsonify( { 'developer': dev } ), 201

This route lives on the same url as the first. But... since won't we just always call the first method since it comes first in the code? The difference here is that, although they live on the same URL, the first method only handles GET requests, while this one only handles POST requests. You can see this in the annotation - methods = ['POST']. Although we're only using one method, you can list any number of HTTP Methods in that list, and if you do, the method following the annotation will handle any request with a matching request method. The first line in the method checks to make sure that the request is actually json, and that a name is passed to the method. If we don't have a name, or it's not json that we're recieving, we return a 400 status code. If we make it past that step, we create a new developer object using the information passed in from the json, add it to the database, and commit the results. Lastly, we return the developer object as json and a 201 status code (HTTP Status Code: Created).

    @app.route('/dev/<int:id>', methods = ['DELETE'])
    def delete_dev(id):
        db.session.delete(Users.query.get(id))
        db.session.commit()
        return jsonify( { 'result': True } )

One of the first things you should notice about the above method is that it handles HTTP Delete methods. Simply enough, this method gets a User by ID and deletes them from the database and commits the database. Finally, it returns {'result':True} to let you know that it completed successfully.

    @app.route('/dev/<int:id>', methods = ['PUT'])
    def update_dev(id):
        dev = Developer.query.get(id)
        dev.name = request.json.get('name', dev.name)
        dev.hireDate = request.json.get('hireDate',dev.name)
        dev.focus = request.json.get('focus', dev.focus)
        db.session.commit()
        return jsonify( { 'dev': dev } )

This last route handles the PUT HTTP Method. Put is the method that is used for updating values, and likewise, our update_dev method is what is called after matching on this route. What this method does is get a developer by their id (primary key) and replaces their attributes with any JSON that is passed in. request.json.get(key, default) is an extremely helpful core-flask method here, that basically says, get the json value located in this key, but if there isn't one, default to the 2nd parameter. In our case, we try to get a name, but we default to the current developer name if we can't find one.

This covers all of our routes, but we have one last piece to add before this will do everything we want it to do. Adding the last statement gives us a nice and easy way to run the application.

    if __name__ == '__main__':
        app.run()

Final Product

If you've been able to follow along up to this point, you should have a final product that looks something like this.

Config.py
    import os
    basedir = os.path.abspath(os.path.dirname(__file__))

    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'app.sqlite')
    SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository')
Flask.py
    from flask import Flask, jsonify, abort, request
    from flask.ext.restless import APIManager
    from flask.ext.sqlalchemy import SQLAlchemy

    app = Flask(__name__)
    app.config.from_pyfile('config.py')
    db = SQLAlchemy(app)

    class Developer(db.Model):
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(20))
        hireDate = db.Column(db.Date)
        focus = db.Column(db.String(50))

        def __init__(self, name, hireDate, focus):
            self.name = name
            self.hireDate = datetime.datetime.strptime(hireDate, "%d%m%Y").date()
            self.focus = focus

    @app.route('/dev/', methods = ['GET'])
    def index():
        return jsonify({'developers': Developer.query.all()})

    @app.route('/dev/<int:id>/')
    def get_dev(id):
        return jsonify({'developer': Developer.query.get(id)})

    @app.route('/dev/', methods = ['POST'])
    def create_dev():
        if not request.json or not 'name' in request.json:
            abort(400)
        dev = Developer(request.json.name, request.json.get('hireDate', ''), request.json.get('focus',''))
        db.session.add(dev)
        db.session.commit()
        return jsonify( { 'developer': dev } ), 201

    @app.route('/dev/<int:id>', methods = ['DELETE'])
    def delete_dev(id):
        db.session.delete(Users.query.get(id))
        db.session.commit()
        return jsonify( { 'result': True } )

    @app.route('/dev/<int:id>', methods = ['PUT'])
    def update_dev(id):
        dev = Developer.query.get(id)
        dev.name = request.json.get('name', dev.name)
        dev.hireDate = request.json.get('hireDate',dev.name)
        dev.focus = request.json.get('focus', dev.focus)
        db.session.commit()
        return jsonify( { 'dev': dev } )

    if __name__ == '__main__':
        app.run()
  • LinkedIn
  • Tumblr
  • Reddit
  • Google+
  • Pinterest
  • Pocket
Comments powered by Disqus