Last active
June 30, 2019 15:46
-
-
Save lotusirous/1f7528dfeb51594610d0f034eab17336 to your computer and use it in GitHub Desktop.
A beautiful example of flask app structure for production in practice
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from flask import Flask | |
from flask import json, Response | |
""" | |
From: https://youtu.be/1ByQhAM5c1I | |
Refer to another repos: | |
- https://github.com/pallets/flask/tree/master/examples/flaskr | |
- https://github.com/wgwz/flask-for-fun-and-profit | |
""" | |
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)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment