Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Flask JSON required decorator
"""
Demo of json_required decorator for API input validation/error handling
"""
import inspect
import functools
import json
from traceback import format_exception
from flask import jsonify, request
import sys
from flask.exceptions import JSONBadRequest
from flask import Flask
import re
app = Flask(__name__)
def api_error_response(code=404, message="Requested resource was not found", errors=list()):
"""
Convenience function for returning a JSON response that includes
appropriate error messages and code.
"""
response = jsonify(dict(code=code, message=message, errors=errors, success=False))
response.status_code = code
return response
def bad_json_error_response():
"""
Convenience function for returning an error message related to
malformed/missing JSON data.
"""
return api_error_response(code=400,
message="There was a problem parsing the supplied JSON data. Please send valid JSON.")
def json_required(func=None, required_fields={}, validations=[]):
"""
Decorator used to validate JSON input to an API request
"""
if func is None:
return functools.partial(json_required, required_fields=required_fields, validations=validations)
@functools.wraps(func)
def decorated_function(*args, **kwargs):
try:
#If no JSON was supplied (or it didn't parse correctly)
try:
if request.json is None:
return bad_json_error_response()
except JSONBadRequest:
return bad_json_error_response()
#Check for specific fields
errors = []
def check_required_fields(data, fields):
for field, requirements in fields.iteritems():
nested_fields = type(requirements) == dict
if data.get(field) in (None, ''):
if nested_fields:
error_msg = requirements.get('message')
else:
error_msg = requirements
errors.append({'field': field, 'message': error_msg})
elif nested_fields:
check_required_fields(data[field], requirements.get('fields', {}))
check_required_fields(request.json, required_fields)
for validation_field, validation_message, validation_func in validations:
func_args = inspect.getargspec(validation_func).args
func_params = []
for arg in func_args:
func_params.append(request.json.get(arg))
if not validation_func(*func_params):
errors.append({'field': validation_field, 'message': validation_message})
if errors:
return api_error_response(code=422, message="JSON Validation Failed", errors=errors)
except Exception:
#For internal use, nice to have the traceback in the API response for debugging
#Probably don't want to include for public APIs
etype, value, tb = sys.exc_info()
error_info = ''.join(format_exception(etype, value, tb))
return api_error_response(code=500, message="Internal Error validating API input", errors=[{'message':error_info}])
return func(*args, **kwargs)
return decorated_function
EMAIL_REGEX = re.compile(r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+)*@(?:[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?\.)+[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?")
def verify_account_available(email):
"""
Check to see if this email is already registered
"""
#Run a query, use an ORM, use Twilio to call someone and ask them :-)
return True
def valid_date_of_birth(date_of_birth):
"""
Does the supplied date string meet our criteria for a date of birth
"""
#Do whatever you need to do...
return True
@app.route("/do/something", methods=['POST'])
@json_required(
required_fields={
'first_name':"Please provide your first name.",
'last_name':"Please provide your last name.",
'email':'Please specify a valid email address',
'date_of_birth':'Please provide your date of birth'
},
validations=[
('email', 'Please provide a valid email address', lambda email: email is not None and EMAIL_REGEX.match(email)),
('email', "This email is already in use. Please try a different email address.", verify_account_available),
('date_of_birth', 'Please provide a valid date of birth', valid_date_of_birth)
]
)
def do_something_useful():
#Confidently use the data in request.json...
return jsonify(dict(status='OK'))
if __name__ == "__main__":
with app.test_client() as client:
response = client.post(
'/do/something',
data=json.dumps({ "first_name": "Brian",
"last_name": "Corbin",
"email": "corbinbs@example.com",
"date_of_birth": "01/01/1970" }),
follow_redirects=True,
content_type='application/json')
response_dict = json.loads(response.data)
assert response_dict['status'] == 'OK'
response = client.post(
'/do/something',
data=json.dumps({ "last_name": "Corbin",
"email": "corbinbs@example.com",
"date_of_birth": "01/01/1970" }),
follow_redirects=True,
content_type='application/json')
response_dict = json.loads(response.data)
assert response.status_code == 422
assert response_dict['code'] == 422
assert response_dict['message'] == "JSON Validation Failed"
assert len(response_dict['errors']) == 1
assert response_dict['errors'][0]['field'] == 'first_name'
assert response_dict['errors'][0]['message'] == 'Please provide your first name.'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.