Überprüfen Sie den veröffentlichten json mit jsonschema, unterbrechen Sie die Verarbeitung, wenn ein Fehlerfeld vorliegt, und antworten Sie mit json mit dem Statuscode 400, in dem der Fehler aufgetreten ist
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