Validez le json publié à l'aide de jsonschema, interrompez le traitement s'il y a un champ d'erreur et répondez avec json avec le code d'état 400 où l'erreur s'est produite
app.py
from flask import Flask
app = Flask(__name__)
import json_schema
import json
schema = {
"name": {
"type": "string",
"required": True
},
"age": {
"type": "number",
"required": True
"maximum": 120,
"minimum": 0
}
}
@app.route("/")
@json_schema.validate("POST", schema)
def index():
if request.is_post:
data = json.loads(request.data)
print data["name"]
print data["age"]
return "Hi! "+data["name"]
else:
return "Hi!"
if __name__ == "__main__":
app.run()
https://gist.github.com/yuiseki/9715d6289ff65fa6e1b2
json_schema.py
import json
from jsonschema import ValidationError, exceptions
from jsonschema.validators import Draft3Validator
from functools import wraps
from flask import _request_ctx_stack, request, jsonify
def _validate(schema, data):
reqv = Draft3Validator(schema)
errors = []
for e in reqv.iter_errors(data):
errors.append(dict(name=e.path[0], reason=e.validator))
return errors
def validate(method, schema):
def decorator(f):
@wraps(f)
def decorated_func(*args, **kws):
ctype = request.headers.get("Content-Type")
method_ = request.headers.get("X-HTTP-Method-Override", request.method)
if method_.lower() == method.lower() and "json" in ctype:
data = json.loads(request.data)
errors = _validate(schema, data)
if len(errors) > 0:
resp = jsonify(result="failure", reason="invalid json", errors=errors)
resp.status_code = 400
return resp
return f(*args, **kws)
return decorated_func
return decorator
Recommended Posts