Flask：验证 JSON 和 JSON Schema 的装饰器

Question

我有一个使用Flask框架的应用程序，它会处理一些需要JSON格式数据的请求。在每次处理请求之前，我会进行两步错误检查：

• 确认接收到的数据是有效的JSON格式
• 确认这个JSON数据符合特定的格式要求

这个检查过程是这样实现的：

@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

因为这段代码在很多请求中都是重复的，所以我在想能不能优雅地把它放到一个装饰器里，类似于这样：

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

问题是，request这个参数是隐含的：我可以在函数内部使用它，但它并不是函数的参数。因此，我不太确定如何在装饰器中使用它。

我该如何使用Python的装饰器来实现这些验证检查呢？

Answer 1

虽然这个回答有点晚，但你可能在找像 marshmallow（flask-marshmallow）或者 toastedmarshmallow 这样的东西。

Answer 2

现在你可以直接使用 @expect_json 这个功能了。

举个例子：

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

或者：

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

想了解更多信息，可以点击 这里。

Answer 3

只需在你的装饰器中使用全局的 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

在使用 @route 装饰器之前，先应用这些装饰器；你需要注册的是被包装过的函数，而不是原始函数，用于路由：

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