在Flask中使用装饰器统一验证接口参数

使用装饰器的方式，拦截请求参数。提前定义好请求参数类，使用marshmallow实现。

import functools

from flask import request
from marshmallow import Schema, ValidationError

from app.response import ResponseCode, error_json
from config import current_config


def validate_schema(schema_class: Schema):
    """验证表单的验证器。会动态设置request.schema_data属性"""

    def decorator(view_func):
        @functools.wraps(view_func)
        def inner(*args, **kwargs):
            if request.method == 'GET':
                form_data = request.args
            else:
                if request.json:
                    form_data = request.json
                else:
                    form_data = request.form
            try:
                data = schema_class().load(form_data)
                request.schema_data = data
            except ValidationError as e:
                if current_config.DEBUG:
                    return error_json(ResponseCode.PARAMETER_ERROR, e.messages)
                else:
                    return error_json(ResponseCode.PARAMETER_ERROR, '参数错误')  # 生产模式的时候,防止恶意用户直接得到参数列表
            return view_func(*args, **kwargs)

        return inner

    return decorator

定义Schema

from marshmallow import fields, validates, validate, ValidationError

from .base_validator import BaseSchema
from .token_validator import TokenSchema


class UserSchema(object):
    password = fields.String(required=True, validate=validate.Length(min=6, max=20))
    username = fields.String(required=True, validate=validate.Length(min=2, max=20))
    tel = fields.String(required=True, validate=validate.Length(min=11, max=11))

    code = fields.String(required=True, validate=validate.Length(min=1))

    _is_openid = fields.Bool(missing=False)
    _access_token = fields.String(missing='')
    _unionid = fields.String(missing='')
    openid = fields.String(required=True)
    unionid = fields.String(required=True)


class PutPasswordSchema(BaseSchema):
    old_password = UserSchema.password
    new_password = UserSchema.password

在视图中使用

@bp.route('/users/password', methods=['PUT'])
@validate_schema(user_valiadtor.PutPasswordSchema)
def put_users_password():
    """用户用户密码
    """
    uid = g.uid
    old_password = request.schema_data.get('old_password')
    new_password = request.schema_data.get('new_password')
    ......