Skip to content

Instantly share code, notes, and snippets.

Last active April 14, 2020 18:45
Show Gist options
  • Save theorm/5440144 to your computer and use it in GitHub Desktop.
Save theorm/5440144 to your computer and use it in GitHub Desktop.
Pluggable API using Flask MethodView.
# -*- coding: utf-8 -*-
from .bananas import Bananas
from .base import the_api
the_api.add_url_rule('/bananas', view_func=Bananas.as_view('bananas'))
the_api.add_url_rule('/farm/<farm_id>/bananas', view_func=Bananas.as_view('bananas_from_a_farm'))
__all__ = [
# -*- coding: utf-8 -*-
from .base import APIEndpoint
from .exceptions import BadRequest
from ..repositories import bananas
from flask import request
class Bananas(APIEndpoint):
def get(self, farm_id=None):
if farm:
return bananas.all(farm_id=farm_id)
return bananas.all()
def post(self, farm_id=None):
'''Create new banana.'''
payload = request.json or {}
banana_type, name = payload.get('type'), payload.get('name')
farm_id = payload.get('farm') or farm_id
if not banana_type or not name or not farm_id:
raise BadRequest('"type", "farm" and "name" are required.')
return, name=name, farm_id=farm_id)
from flask import Blueprint, make_response, json, current_app
from bson import ObjectId
from flask.views import MethodView
from ..pymongo import DuplicateKeyError
from .exceptions import BadRequest, Unauthorized, Forbidden
from import get_current_user
the_api = Blueprint('the_api', __name__)
class JsonEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, ObjectId):
return str(obj)
return json.JSONEncoder.default(self, obj)
def handle_exception(e):
# handle common exceptions
if isinstance(e, DuplicateKeyError):
raise BadRequest(str(e))
def json_converter(f):
'''Converts `dict`, list or mongo cursor to JSON.
Creates `~flask.Response` object and sets headers.
def decorator(*args, **kwargs):
result = f(*args, **kwargs)
except Exception as e:
if isinstance(result, dict):
result = json.dumps(result, cls=JsonEncoder)
# unwind cursor
result = json.dumps(list(result), cls=JsonEncoder)
response = make_response(result)
response.headers['Content-Type'] = 'application/json; charset=utf-8'
return response
return decorator
def login_required(f):
def decorator(*args, **kwargs):
if not get_current_user().is_authenticated():
raise Unauthorized('You must log in to access this URL.')
return f(*args, **kwargs)
return decorator
def admin_required(f):
def decorator(*args, **kwargs):
if not get_current_user().is_admin():
raise Forbidden('You must be an admin to access this URL.')
return f(*args, **kwargs)
return decorator
class APIEndpoint(MethodView):
# make converter run after every request handler method returns
decorators = [json_converter, login_required]
def __init__(self, *args, **kwargs):
super(APIEndpoint, self).__init__(*args, **kwargs)
self.logger = current_app.logger
class AdminAPIEndpoint(APIEndpoint):
decorators = APIEndpoint.decorators + [admin_required]
# -*- coding: utf-8 -*-
Wrapper classes to return JSON exception, not HTML exception as werkzeug does.
from flask.exceptions import JSONHTTPException
from werkzeug import exceptions
class BadRequest(JSONHTTPException, exceptions.BadRequest):
class Unauthorized(JSONHTTPException, exceptions.Unauthorized):
class Forbidden(JSONHTTPException, exceptions.Forbidden):
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment