The examples below will use httpie (a curl-like tool) for testing the APIs.

Text Analysis API (Bottle + TextBlob)

Here is a very simple text analysis API using Bottle and TextBlob that demonstrates how to declare an object serializer.

Assume that TextBlob objects have polarity, subjectivity, noun_phrase, tags, and words properties.

from bottle import route, request, run
from textblob import TextBlob
from marshmallow import Schema, fields

class BlobSchema(Schema):
    polarity = fields.Float()
    subjectivity = fields.Float()
    chunks = fields.List(fields.String, attribute="noun_phrases")
    tags = fields.Raw()
    discrete_sentiment = fields.Method("get_discrete_sentiment")
    word_count = fields.Function(lambda obj: len(obj.words))

    def get_discrete_sentiment(self, obj):
        if obj.polarity > 0.1:
            return 'positive'
        elif obj.polarity < -0.1:
            return 'negative'
            return 'neutral'

blob_schema = BlobSchema()

@route("/api/v1/analyze", method="POST")
def analyze():
    blob = TextBlob(request.json['text'])
    result = blob_schema.dump(blob)
    return result.data

run(reloader=True, port=5000)

Using The API

First, run the app.

$ python textblob_example.py

Then send a POST request with some text.

$ http POST :5000/api/v1/analyze text="Simple is better"
HTTP/1.0 200 OK
Content-Length: 189
Content-Type: application/json
Date: Wed, 13 Nov 2013 08:58:40 GMT
Server: WSGIServer/0.1 Python/2.7.5

    "chunks": [
    "discrete_sentiment": "positive",
    "polarity": 0.25,
    "subjectivity": 0.4285714285714286,
    "tags": [
    "word_count": 3

Quotes API (Flask + SQLAlchemy)

Below is a full example of a REST API for a quotes app using Flask and SQLAlchemy with marshmallow. It demonstrates a number of features, including:

  • class Meta to specify which fields to serialize
  • Nesting fields
  • Output filtering using the only parameter
from datetime import datetime

from flask import Flask, jsonify, request
from flask.ext.sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError
from marshmallow import Schema, fields

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = 'sqlite:////tmp/quotes.db'
db = SQLAlchemy(app)

##### MODELS #####

class Author(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first = db.Column(db.String(80))
    last = db.Column(db.String(80))

    def __init__(self, first, last):
        self.first = first
        self.last = last

class Quote(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String, nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("author.id"))
    author = db.relationship("Author",
                        backref=db.backref("quotes", lazy="dynamic"))
    posted_at = db.Column(db.DateTime)

    def __init__(self, content, author):
        self.author = author
        self.content = content
        self.posted_at = datetime.utcnow()

##### SCHEMAS #####

class AuthorSchema(Schema):
    formatted_name = fields.Method("format_name")

    def format_name(self, author):
        return "{}, {}".format(author.last, author.first)

    class Meta:
        fields = ('id', 'first', 'last', "formatted_name")

class QuoteSchema(Schema):
    author = fields.Nested(AuthorSchema)

    class Meta:
        fields = ("id", "content", "posted_at", 'author')

author_serializer = AuthorSchema()
quote_serializer = QuoteSchema()
quotes_serializer = QuoteSchema(many=True, only=('id', 'content'))

##### API #####

def get_authors():
    authors = Author.query.all()
    # Serialize the queryset
    serializer = AuthorSchema(many=True)
    result = serializer.dump(authors)
    return jsonify({"authors": result.data})

def get_author(pk):
        author = Author.query.get(pk)
    except IntegrityError:
        return jsonify({"message": "Author could not be found."}), 400
    author_result = author_serializer.dump(author)
    quotes_result = quotes_serializer.dump(author.quotes.all())
    return jsonify({'author': author_result.data, 'quotes': quotes_result.data})

@app.route('/api/v1/quotes', methods=['GET'])
def get_quotes():
    quotes = Quote.query.all()
    result = quotes_serializer.dump(quotes)
    return jsonify({"quotes": result.data})

def get_quote(pk):
        quote = Quote.query.get(pk)
    except IntegrityError:
        return jsonify({"message": "Quote could not be found."}), 400
    result = quote_serializer.dump(quote)
    return jsonify({"quote": result.data})

@app.route("/api/v1/quotes/new", methods=["POST"])
def new_quote():
    first, last = request.json['author'].split(" ")
    content = request.json['quote']
    author = Author.query.filter_by(first=first, last=last).first()
    if author is None:
        # Create a new author
        author = Author(first, last)
    # Create new quote
    quote = Quote(content, author)
    result = quote_serializer.dump(Quote.query.get(quote.id))
    return jsonify({"message": "Created new quote.",
                    "quote": result.data})

if __name__ == '__main__':
    app.run(debug=True, port=5000)

Using The API

Run the app.

$ python flask_example.py

First we’ll POST some quotes.

$ http POST :5000/api/v1/quotes/new author="Tim Peters" quote="Beautiful is better than ugly."
$ http POST :5000/api/v1/quotes/new author="Tim Peters" quote="Now is better than never."
$ http POST :5000/api/v1/quotes/new author="Peter Hintjens" quote="Simplicity is always better than functionality."

Now we can GET a list of all the quotes.

$ http :5000/api/v1/quotes
    "quotes": [
            "content": "Beautiful is better than ugly.",
            "id": 1
            "content": "Now is better than never.",
            "id": 2
            "content": "Simplicity is always better than functionality.",
            "id": 3

We can also GET the quotes for a single author.

$ http :5000/api/v1/authors/1
    "author": {
        "first": "Tim",
        "formatted_name": "Peters, Tim",
        "id": 1,
        "last": "Peters"
    "quotes": [
            "content": "Beautiful is better than ugly.",
            "id": 1
            "content": "Now is better than never.",
            "id": 2

ToDo API (Flask + Peewee)

This example uses Flask and the Peewee ORM to create a basic Todo application.

Notice how __marshallable__ is used to define how Peewee model objects get marshalled.

import datetime as dt
from functools import wraps

from flask import Flask, request, g, jsonify
import peewee as pw
from marshmallow import Schema, fields

app = Flask(__name__)
db = pw.SqliteDatabase("/tmp/todo.db")

###### MODELS #####

class BaseModel(pw.Model):
    """Base model class. All descendants share the same database."""
    def __marshallable__(self):
        """Return the marshallable dictionary that will be serialized by
        marshmallow. Peewee models have a dictionary representation where the
        ``_data`` key contains all the field:value pairs for the object.
        return dict(self.__dict__)['_data']

    class Meta:
        database = db

class User(BaseModel):
    email = pw.CharField(max_length=80, unique=True)
    password = pw.CharField()
    joined_on = pw.DateTimeField()

class Todo(BaseModel):
    content = pw.TextField()
    is_done = pw.BooleanField(default=False)
    user = pw.ForeignKeyField(User)
    posted_on = pw.DateTimeField()

    class Meta:
        order_by = ('-posted_on', )

def create_tables():

##### SCHEMAS #####

class UserSchema(Schema):
    class Meta:
        fields = ('email', 'joined_on')

class TodoSchema(Schema):
    done = fields.Boolean(attribute='is_done')
    user = fields.Nested(UserSchema)

    class Meta:
        additional = ('id', 'content', 'posted_on')

    def make_object(self, data):
        user = User.get(User.email == data['user'])
        if data.get('id'):
            todo = Todo.get(Todo.id == data.get['id'])
            todo = Todo(content=data['content'],
                        posted_on=data.get('posted_on') or dt.datetime.utcnow())
        return todo

user_serializer = UserSchema()
todo_serializer = TodoSchema()
todos_serializer = TodoSchema(many=True)

###### HELPERS ######

def check_auth(email, password):
    """Check if a username/password combination is valid.
        user = User.get(User.email == email)
    except User.DoesNotExist:
        return False
    return password == user.password

def requires_auth(f):
    def decorated(*args, **kwargs):
        auth = request.authorization
        if not auth or not check_auth(auth.username, auth.password):
            resp = jsonify({"message": "Please authenticate."})
            resp.status_code = 401
            resp.headers['WWW-Authenticate'] = 'Basic realm="Example"'
            return resp
        return f(*args, **kwargs)
    return decorated

#### API #####

# Ensure a separate connection for each thread
def before_request():
    g.db = db

def after_request(response):
    return response

@app.route("/api/v1/register", methods=["POST"])
def register():
    try:  # Use get to see if user already to exists
        User.get(User.email == request.json['email'])
        message = "That email address is already in the database."
    except User.DoesNotExist:
        user = User.create(email=request.json['email'], joined_on=dt.datetime.now(),
        message = "Successfully created user: {0}".format(user.email)
    data, errors = user_serializer.dump(user)
    if errors:
        return jsonify(errors), 400
    return jsonify({'message': message, "user": data})

def get_todos():
    todos = Todo.select()  # Get all todos
    data, errors = todos_serializer.dump(list(todos))
    if errors:
        return jsonify(errors), 400
    return jsonify({"todos": data})

def get_todo(pk):
        todo, errs = todo_serializer.load({'id': pk})
    except Todo.DoesNotExist:
        return jsonify({"message": "Todo could not be found"})
    data, errors = todo_serializer.dump(todo)
    if errors:
        return jsonify(errors), 400
    return jsonify({"todo": data})

@app.route("/api/v1/todos/<int:pk>/toggle", methods=["POST"])
def toggledone(pk):
        todo = Todo.get(Todo.id == pk)
    except Todo.DoesNotExist:
        return jsonify({"message": "Todo could not be found"})
    status = not todo.is_done
    update_query = todo.update(is_done=status)
    data, errors = todo_serializer.dump(todo)
    if errors:
        return jsonify(errors), 400
    return jsonify({"message": "Successfully toggled status.",
                    "todo": data})

@app.route("/api/v1/todos/new", methods=["POST"])
def new_todo():
    todo, errs = todo_serializer.load({
        'content': request.json['content'],
        'user': request.authorization.username,
        'posted_on': dt.datetime.now(),
    data, errors = todo_serializer.dump(todo)
    if errors:
        return jsonify(errors), 400
    return jsonify({"message": "Successfully created new todo item.",
                    "todo": data})

if __name__ == '__main__':
    app.run(port=5000, debug=True)

Using the API

After registering a user and creating some todo items in the database, here is an example response.

$ http GET :5000/api/v1/todos
    "todos": [
            "content": "Refactor everything",
            "done": false,
            "id": 3,
            "posted_on": ""2014-08-17T14:42:12.479650+00:00",
            "user": {
                "email": "foo@bar.com",
                "joined_on": "2014-08-14T13:12:19.179650+00:00"
            "content": "Learn python",
            "done": false,
            "id": 2,
            "posted_on": "2014-08-15T17:41:12.479650+00:00",
            "user": {
                "email": "foo@bar.com",
                "joined_on": "2014-08-14T13:12:19.179650+00:00"
            "content": "Install marshmallow",
            "done": false,
            "id": 1,
            "posted_on": "2014-08-15T09:15:12.479650+00:00",
            "user": {
                "email": "foo@bar.com",
                "joined_on": "2014-08-14T13:12:19.179650+00:00"

Object serialization and deserialization, lightweight and fluffy.

Useful Links

Table Of Contents

Related Topics

Fork me on GitHub