Flask: Decorator to verify JSON and JSON Schema
Just use the request
context global in your decorator. It is available during any request.
from functools import wraps
from flask import (
current_app,
jsonify,
request,
)
def validate_json(f):
@wraps(f)
def wrapper(*args, **kw):
try:
request.json
except BadRequest, e:
msg = "payload must be a valid json"
return jsonify({"error": msg}), 400
return f(*args, **kw)
return wrapper
def validate_schema(schema_name):
def decorator(f):
@wraps(f)
def wrapper(*args, **kw):
try:
validate(request.json, current_app.config[schema_name])
except ValidationError, e:
return jsonify({"error": e.message}), 400
return f(*args, **kw)
return wrapper
return decorator
Apply these decorators before applying the @route
decorator; you want to register the wrapped function, not the original function for the route:
@app.route('/activate', methods=['POST'])
@validate_json
@validate_schema('activate_schema')
def activate():
input = request.json
now you can use @expect_json
directly
For Example
from flask import Flask, jsonify, g, url_for
from flask_expects_json import expects_json
# example imports
from models import User
from orm import NotUniqueError
app = Flask(__name__)
schema = {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'email': {'type': 'string'},
'password': {'type': 'string'}
},
'required': ['email', 'password']
}
@app.route('/register', methods=['POST'])
@expects_json(schema)
def register():
# if payload is invalid, request will be aborted with error code 400
# if payload is valid it is stored in g.data
# do something with your data
user = User().from_dict(g.data)
try:
user.save()
except NotUniqueError as e:
# exception path: duplicate database entry
return jsonify(dict(message=e.message)), 409
# happy path: json response
resp = jsonify(dict(auth_token=user.encode_auth_token(), user=user.to_dict()})
resp.headers['Location'] = url_for('users.get_user', user_id=user.id)
return resp, 201
or
from flask import Flask
from flask_expects_json import expects_json
app = Flask(__name__)
schema = {
'type': 'object',
'properties': {
'name': {'type': 'string', "minLength": 4, "maxLength": 15},
'mobile': {'type': 'string', "pattern": "^[1-9]{1}[0-9]{9}$"},
'email': {'type': 'string', "pattern": "[^@]+@[^@]+\.[^@]"},
'password': {'type': 'string', "pattern": "^.*(?=.{8,})(?=.*\d)(?=.*[a-z])(?=.*[A-Z])(?=.*[!@#$%^&+=]).*$"}
},
'required': ['name', 'mobile', 'email', 'password']
}
@app.route('/', methods=['POST'])
@expects_json(schema)
def index():
values = request.get_json()
print(values)
return values
get more from here