Flask: Decorator to verify JSON and JSON Schema

Question

I have a flask application with calls expecting JSON payload. Before each call is processed, I have a 2-step error checking process:

• Assert that the payload is a valid JSON
• Assert that the JSON payload complies with a specific schema

Which is implemented in the following fashion:

@app.route('/activate', methods=['POST']) def activate():     request_id = request.__hash__()      # Assert that the payload is a valid JSON     try:         input = request.json     except BadRequest, e:         msg = "payload must be a valid json"         return jsonify({"error": msg}), 400      # JSON Schema Validation     try:         validate(request.json, app.config['activate_schema'])     except ValidationError, e:         return jsonify({"error": e.message}), 400 

Since this code is duplicated over many calls, I wonder If I can elegantly move it to a decorator, something in the formof:

@validate_json @validate_schema(schema=app.config['activate_schema']) @app.route('/activate', methods=['POST']) def activate():     .... 

The problem is that the request argument is implicit: I can refer to it within the function, but it is not a parameter to it. Therefore, I am not sure how to use it within the decorator.

How can I implement the validation checks using Python decorators?

Answer 1

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 

Answer 2

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