Simple Flask REST API

flask_rest_api.py

from flask import Flask
from flask import json, Response

"""
From: https://youtu.be/1ByQhAM5c1I
"""


class ApiResult(object):
    """
    API Result Wrapper

    * Consistent
    * Speak same type of JSON
    * Similar response/headers/mime type
    Stores return value we want to generate

    Also does pagination
    """

    def __init__(self, value, status=200):
        self.value = value
        self.status = status

    def to_response(self):
        return Resonse(json.dumps(self.value),
                       status=self.status,
                       mimetype='application/json')


class ApiFlask(Flask):
    """
    Response Converter
    """

    def make_response(self, rv):
        if instance(rv, ApiResult):
            return rv.to_response()
        return Flask.make_response(self, rv)


class ApiException(object):
    """
    API Errors
    """

    def __init__(self, message, status=400)
        self.message = message
        self.status = status

    def to_result(self):
        return ApiResult({'message': self.message},
                         status=self.status)


def register_error_handlers(app):
    app.register_error_handler(
            ApiException, lambda err: err.to_result())


register_error_handlers(app)

"""
Example
"""

from flask import BluePrint

bp = Blueprint('demo', __name__)

@bp.route('/add')
def add_numbers():
    a = request.args.get('a', type=int)
    b = request.args.get('b', type=int)
    if a is None or b is None:
        raise ApiException('Numbers must be integers')
    return ApiResult({'sum': a + b})

#jsonschema
#voluptuous


from flask import request
from voluptuous import Invalid


def dataschema(schema):
    def decorator(f):
        def new_func(*args, **kwargs):
            try:
                kwargs.update(schema(request.get_json()))
            except Invalid as e:
                raise ApiException('Invalid data: %s (path "%s")' %
                                   (e.msg, '.'.join(e.path)))
            return f(*args, **kwargs)
        return update_wrapper(new_func, f)
    return decorator

from voluptous import Schema, REMOVE_EXTRA


@app.route('/add', methods=['POST'])
@dataschema(Schema({
    'a': int,
    'b': int,
}, extra=REMOVE_EXTRA)
def add_numbers(a, b):
    return ApiResult({'sum': a + b})

            
# Control the API: Pagination


from werkeug.urls import url_join

class ApiResult(object):

    def  __init__(self, ..., next_page=None):
        self.next_page = next_page

    def to_response(self):
        rv = Response()
        if self.next_page is not None:
            rv.headers['Link'] = '<%s>; rel="next"' % \
                    url_join(request.url, self.next_page)
        return rv

# Security

from myapp import db
from myapp.security import get_available_organizations

class Project(db.model):

    @property
    def query(self):
        org_query = get_available_organizations()
        return db.Query(self).filter(
            Project.organization.in_(org_query))