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
  • Validation using Schema.validate().
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, ValidationError

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")

def must_not_be_blank(data):
    if not data:
        raise ValidationError('Data not provided.')

class QuoteSchema(Schema):
    author = fields.Nested(AuthorSchema, validate=must_not_be_blank)
    content = fields.Str(required=True, validate=must_not_be_blank)

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

author_schema = AuthorSchema()
quote_schema = QuoteSchema()
quotes_schema = 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_schema.dump(author)
    quotes_result = quotes_schema.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_schema.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_schema.dump(quote)
    return jsonify({"quote": result.data})

@app.route("/api/v1/quotes/", methods=["POST"])
def new_quote():
    if not request.get_json():
        return jsonify({'message': 'No input data provided'}), 400
    author_name = request.get_json().get('author')
    if author_name:
        first, last = author_name.split(' ')
        author_input = dict(first=first, last=last)
        author_input = {}
    content_input = request.get_json().get('content')
    input_data = dict(author=author_input, content=content_input)
    # Validate the input data
    errors = quote_schema.validate(input_data)
    if errors:
        return jsonify(errors), 400
    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_input, author)
    result = quote_schema.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/ author="Tim Peters" content="Beautiful is better than ugly."
$ http POST :5000/api/v1/quotes/ author="Tim Peters" content="Now is better than never."
$ http POST :5000/api/v1/quotes/ author="Peter Hintjens" content="Simplicity is always better than functionality."

If we provide invalid input data, we get 400 error response. Let’s omit “author” from the input data.

$ http POST :5000/api/v1/quotes/ content="I have no author"
    "author": [
        "Data not provided."

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. We also use the Schema.load() method to deserialize input data to an ORM object (see the new_todo() view).

import datetime as dt
from functools import wraps

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

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 = ('id', 'email', 'joined_on')

class TodoSchema(Schema):
    done = fields.Boolean(attribute='is_done')
    user = fields.Nested(UserSchema, exclude=('joined_on',))
    content = fields.Str(required=True)
    posted_on = fields.DateTime()

    class Meta:
        additional = ('id', )

    def make_object(self, data):
        if data.get('id'):
            todo = Todo.get(Todo.id == data['id'])
            user_data = data.get('user')
            if not user_data:
                raise ValidationError('Must provide user when creating a new todo item.')
            user = User.get(User.email == user_data['email'])
            todo = Todo(content=data['content'],
        return todo

user_schema = UserSchema()
todo_schema = TodoSchema()
todos_schema = 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():
    json_input = request.get_json()
    if not json_input:
        return jsonify({'message': 'Must provide username and password'}), 400
    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=json_input['email'], joined_on=dt.datetime.now(),
        message = "Successfully created user: {0}".format(user.email)
    data, errors = user_schema.dump(user)
    if errors:
        return jsonify(errors), 400
    return jsonify({'message': message, "user": data})

@app.route("/api/v1/todos/", methods=['GET'])
def get_todos():
    todos = Todo.select()  # Get all todos
    data, errors = todos_schema.dump(list(todos))
    if errors:
        return jsonify(errors), 400
    return jsonify({"todos": data})

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

@app.route("/api/v1/todos/<int:pk>/toggle", methods=["POST", "PUT"])
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_schema.dump(todo)
    if errors:
        return jsonify(errors), 400
    return jsonify({"message": "Successfully toggled status.",
                    "todo": data})

@app.route("/api/v1/todos/", methods=["POST"])
def new_todo():
    json_input = request.get_json()
    if not json_input:
        return jsonify({'message': 'No input provided'}), 400
    # Deserialize to a new Todo
    todo, errors = todo_schema.load({
        'content': json_input.get('content'),
        'user': {'email': request.authorization.username}
    if errors:
        return jsonify(errors), 400
    result = todo_schema.dump(todo)
    return jsonify({"message": "Successfully created new todo item.",
                    "todo": result.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": "Install marshmallow",
            "done": false,
            "id": 3,
            "posted_on": "2014-12-02T02:58:14.070877+00:00",
            "user": {
                "email": "foo@bar.com",
                "id": 1
            "content": "Learn Python",
            "done": false,
            "id": 2,
            "posted_on": "2014-12-02T02:58:08.910516+00:00",
            "user": {
                "email": "foo@bar.com",
                "id": 1
            "content": "Refactor everything",
            "done": false,
            "id": 1,
            "posted_on": "2014-12-02T02:58:04.207961+00:00",
            "user": {
                "email": "foo@bar.com",
                "id": 1

Object serialization and deserialization, lightweight and fluffy.

Useful Links

Table Of Contents

Related Topics

Fork me on GitHub