Python 参数校验的进化

悠悠 2022-04-03 12:48 241阅读 0赞

事情的起因是感觉目前项目中的参数校验方法写的太简单了,很多时候需要在server层再if else处理,于是就动手准备写一个好用一点的,可以自定义校验参数规则的参数校验器,考虑到要可以灵活的配置就萌生了大概的印象:

1.  使用map - 参数A:ruleA,参数B-ruleB..等等,对参数进行规则绑定
2.  [Python装饰器][Python]
3.  可扩展,可以自定义校验规则

于是第一个版本实现如下:

### 版本1 ###

# -*- coding:utf-8 -*-
    __author__ = "aleimu"
    __date__ = "2018-12-6"
    __doc__ = "一个实用的入参校验装饰器--针对目前,前端 url?&a=1&b=2或-d'a=1&b=2c=qwe'形式的非json(所有参数都是str类型)" \
              "入参的校验"
    
    import copy
    import traceback
    from collections import OrderedDict
    from functools import wraps
    from flask import Flask, json, jsonify, request
    
    app = Flask(__name__)
    
    
    def verify_args(need=None, length=None, check=None, strip=True, default=(False, None), diy_func=None, release=False):
        """
        约束:
        1. 简化了传参校验,使用位置传参或者关键词传参(一个参数对应一个参数),不允许使用one to list等python高级传参特性
        2. 所有的参数都是str/unicode类型的,前端没有使用json带参数类型的入参方式
        :param need: 必须参数,且不能为None或者""
        :param length: 参数长度范围
        :param check:  str的常用类方法/属性如下:
            isalnum 判断字符串中只能由字母和数字的组合，不能有特殊符号
            isalpha 字符串里面都是字母，并且至少是一个字母，结果就为真，（汉字也可以）其他情况为假
            isdigit 函数判断是否全为数字
        :param strip:对字段进行前后过滤空格
        :param default:将"" 装换成None
        :param diy_func:自定义的对某一参数的校验函数格式: {key:func},类似check, diy_func={"a": lambda x: x + "aa"})
        :param release:发生参数校验异常后是否依然让参数进入主流程函数
        :return:
        """
    
        def wraps_1(f):
            @wraps(f)
            def wraps_2(*args, **kwargs):
                if release:
                    args_bak = args[:]
                    kwargs_bak = copy.deepcopy(kwargs)  # 下面流程异常时,是否直接使用 原参数传入f todo
                print ("in", args, kwargs)
                args_template = f.func_code.co_varnames
                print("args_template:", args_template)
                args_dict = OrderedDict()
                req_args_need_list = []
                req_args_types_list = []
                try:
                    for i, x in enumerate(args):
                        args_dict[args_template[i]] = x
                    sorted_kwargs = sort_by_co_varnames(args_template, kwargs)
                    args_dict.update(sorted_kwargs)
                    print("args_dict:", args_dict)
                    # need
                    if need:
                        for k in need:
                            if k not in args_dict:
                                req_args_need_list.append(k)
                            else:
                                if args_dict[k] == None or args_dict[k] == "":
                                    req_args_need_list.append(k)
                        if req_args_need_list:
                            return False, "%s is in need" % req_args_need_list
                    # strip
                    if strip:
                        for k in args_dict:
                            if args_dict[k]:
                                args_dict[k] = args_dict[k].strip()
                    # length
                    if length:
                        for k in args_dict:
                            if k in length:
                                if not (len(args_dict[k]) >= length[k][0] and len(args_dict[k]) <= length[k][1]):
                                    return False, "%s length err" % k
                    # default:
                    if default[0]:
                        for x in args_dict:
                            if args_dict[x] == "":
                                args_dict[x] = default[1]
                    # check
                    if check:
                        for k in check:
                            check_func = getattr(type(args_dict[k]), check[k], None)
                            if not (k in args_dict and check_func and check_func(args_dict[k])):
                                req_args_types_list.append(k)
                        if req_args_types_list:
                            return False, "%s type err" % req_args_types_list
                    # diy_func
                    if diy_func:
                        for k in args_dict:
                            if k in diy_func:
                                args_dict[k] = diy_func[k](args_dict[k])
                except Exception as e:
                    print("verify_args catch err: ", traceback.format_exc())
                    if release:
                        return f(*args_bak, **kwargs_bak)
                    else:
                        return False, str(e)
                return f(*args_dict.values())
    
            return wraps_2
    
        return wraps_1
    
    
    def sort_by_co_varnames(all_args, kwargs):
        new_ordered = OrderedDict()
        for x in all_args:
            if x in kwargs:
                new_ordered[x] = kwargs[x]
        return new_ordered
    
    
    @app.route("/", methods=["GET", "POST", "PUT"])
    def index():
        a = request.values.get("a")
        b = request.values.get("b")
        c = request.values.get("c")
        d = request.values.get("d")
        e = request.values.get("e")
        f = request.values.get("f")
        g = request.values.get("g")
        status, data = todo(a, b, c, d, e=e, f=f, g=g)
        if status:
            return jsonify({"code": 200, "data": data, "err": None})
        else:
            return jsonify({"code": 500, "data": None, "err": data})
    
    
    @verify_args(need=['a', 'b', 'c'], length={"a": (6, 50)}, strip=True,
                 check={"b": 'isdigit', "c": "isalnum"},
                 default=(True, None),
                 diy_func={"a": lambda x: x + "aa"})
    def todo(a, b, c, d, e='  1  ', f='2    ', g=''):
        return True, {"a": a, "b": b, "c": c, "d": d, "e": e, "f": f, "g": g}
    
    
    if __name__ == "__main__":
        app.run(host='0.0.0.0', port=6000, debug=True)
    
    """
    # curl "http://127.0.0.1:6000/" -d "pwd=123&a=1111111&b=2&c=3&d=d&e=eeeeee&f=12345&g="
    {
      "code": 200,
      "data": {
        "a": "1111111aa",
        "b": "2",
        "c": "3",
        "d": "d",
        "e": "eeeeee",
        "f": "12345",
        "g": null
      },
      "err": null
    }
    
    # curl "http://127.0.0.1:6000/" -d "pwd=123&a=1111111&b=2&c=3346()*&d=d&e=eeeeee&f=12345&g="
    {
      "code": 500,
      "data": null,
      "err": "['c'] type err"
    }
    
    # curl "http://127.0.0.1:6000/" -d "pwd=123&a=1111111&b=2&c=&d=d&e=eeeeee&f=12345&g="    
    {                                                                                        
      "code": 500,                                                                           
      "data": null,                                                                          
      "err": "['c'] is in need"                                                              
    }   
    
    # curl "http://127.0.0.1:6000/" -d "pwd=123&a=1111111&b=2&c=  1  &d=d&e=eeeeee&f=12345&g="  
    {                                                                                           
      "code": 200,                                                                              
      "data": {                                                                                 
        "a": "1111111aa",                                                                       
        "b": "2",                                                                               
        "c": "1",                                                                               
        "d": "d",                                                                               
        "e": "eeeeee",                                                                          
        "f": "12345",                                                                           
        "g": null                                                                               
      },                                                                                        
      "err": null                                                                               
    }                                                                                                                                                                                
    """

第一个版本切合了当前项目中经常遇到的校验问题,实现起来较简单,基本满足要求.  
想要更通用点,更多校验规则一些,就需要每次为verify\_args添加参数写if else了,嗯.....有点不优雅啊,于是去看github上有啥好的实现.  
找到了如下几个项目:

1.  [https://github.com/keleshev/s...][https_github.com_keleshev_s...] 嗯,1.6K的star,思路一致,实现的优雅,但是不好扩展啊....
2.  [https://github.com/kvesteri/v...][https_github.com_kvesteri_v...] 额,Python Data Validation for Humans™. not for me....
3.  [https://github.com/mansam/val...][https_github.com_mansam_val...] 嗯,思路一致,实现也简单,挺好扩展的,就用它了!

这里说说validator.py ,给个例子

from validator import Required, Not, Truthy, Blank, Range, Equals, In, validate
    
    # let's say that my dictionary needs to meet the following rules...
    rules = {
        "foo": [Required, Equals(123)],
        "bar": [Required, Truthy()],
        "baz": [In(["spam", "eggs", "bacon"])],
        "qux": [Not(Range(1, 100))] # by default, Range is inclusive
    }
    
    # then this following dict would pass:
    passes = {
        "foo": 123,
        "bar": True, # or a non-empty string, or a non-zero int, etc...
        "baz": "spam",
        "qux": 101
    }
    print validate(rules, passes)
    # (True, {})
    
    # but this one would fail
    fails = {
        "foo": 321,
        "bar": False, # or 0, or [], or an empty string, etc...
        "baz": "barf",
        "qux": 99
    }
    print validate(rules, fails)
    # (False,
    #  {
    #  'foo': ["must be equal to '123'"],
    #  'bar': ['must be True-equivalent value'],
    #  'baz': ["must be one of ['spam', 'eggs', 'bacon']"],
    #  'qux': ['must not fall between 1 and 100']
    #  })

嗯,使用第一个版本封装一下validator.py就好了!考虑到需要写个dome来试试,就选了flask,嗯,对了,先去github 上搜一下 flask validator 没准已经有现成的呢,实现思路基本一致,但是......前几个star多的都不令人满意,还是自己造轮子吧.  
先实现常见的在route上加装饰器版本,这样的话,就可以直接接收request收到的参数,然后直接校验了,有问题就直接返回错误给调用者,于是有了版本2

### 版本2 ###

rules_example = {
        "a": [Required, Equals("123")],  # foo must be exactly equal to 123
        "b": [Required, Truthy()],  # bar must be equivalent to True
        "c": [In(["spam", "eggs", "bacon"])],  # baz must be one of these options
        "d": [Not(Range(1, 100))],  # qux must not be a number between 1 and 100 inclusive
        "e": [Length(0, maximum=5)],
        "f": [Required, InstanceOf(str)],
        "g": [Required, Not(In(["spam", "eggs", "bacon"]))],
        "h": [Required, Pattern("\d\d\%")],
        "i": [Required, GreaterThan(1, reverse=True, auto=True)],  # auto 自动转换成float类型来做比较
        "j": [lambda x: x == "bar"],
        "k": [Required, Isalnum()],  # 判断字符串中只能由字母和数字的组合，不能有特殊符号
        "l": [Required, Isalpha()],  # 字符串里面都是字母，并且至少是一个字母，结果就为真，（汉字也可以）其他情况为假
        "m": [Required, Isdigit()],  # 判断字符串是否全为数字
    }
    
    
    def validator_wrap(rules, strip=True, diy_func=None):
        """装饰器版 - 只能检测是否符合规则,不能修改参数
        :param rules:参数的校验规则,map
        :param strip:对字段进行前后空格检测
        :param diy_func:自定义的对某一参数的校验函数格式: {key:func},类似check, diy_func={"a": lambda x: x=="aa"})
        """
    
        def decorator(f):
            @wraps(f)
            def decorated_func(*args, **kwargs):
                try:
                    args_dict = OrderedDict()
                    if request.values:
                        args_dict.update(request.values)
                    if request.json:
                        args_dict.update(request.json)
                    # strip
                    if strip:
                        for k in args_dict:
                            if args_dict[k] and isstr(args_dict[k]):
                                if args_dict[k][0] == " " or args_dict[k][-1] == " ":
                                    return jsonify({"code": 500, "data": None, "err": "%s should not contain spaces" % k})
                    # diy_func
                    if diy_func:
                        for k in args_dict:
                            if k in diy_func:
                                args_dict[k] = diy_func[k](args_dict[k])
                    # rules
                    if rules:
                        result, err = validate(rules, args_dict)
                        if not result:
                            return jsonify(
                                {"code": 500, "data": None, "err": err})
                except Exception as e:
                    print("verify_args catch err: ", traceback.format_exc())
                    return jsonify({"code": 500, "data": None, "err": str(e)})
                return f(*args, **kwargs)
    
            return decorated_func
    
        return decorator
        
    @app.route("/wrap", methods=["GET", "POST", "PUT"])
    @validator_wrap(rules=rules_example, strip=True)  # 姿势 1:只能检测是否符合规则,不能修改参数,不符合就会直接返回json给调用者
    def wrap_example():
        a = request.values.get("a")
        b = request.values.get("b")
        c = request.values.get("c")
        d = request.values.get("d")
        e = request.values.get("e")
        f = request.values.get("f")
        g = request.values.get("g")
        h = request.values.get("h")
        i = request.values.get("i")
        j = request.values.get("j")
        k = request.values.get("k")
        l = request.values.get("l")
        m = request.values.get("m")
        status, data = todo(a=a, b=b, c=c, d=d, e=e, f=f, g=g, h=h, i=i, j=j, k=k, l=l, m=m)
        if status:
            return jsonify({"code": 200, "data": data, "err": None})
        else:
            return jsonify({"code": 500, "data": None, "err": data})

好像挺好的,基本满足要求了,但是再route上加装饰器,那就改变不了参数的值了,虽然有些参数不一定符合要求,但是简单修补一下还是可以用的,还得继续寻找能够改变入参的方式,第一反应是在装饰器中修改request.values或者request.json的值,让进入到主函数后获取更新后的值,上下求索未得门径,request.value.update方法是被禁用的,继续看源码,后面的实现使用了dict的复杂封装,不好改啊,这样太绕了,还是直接调用函数吧,不玩装饰器了.于是又了版本3

### 版本3 ###

def validator_func(rules, strip=True, default=(False, None), diy_func=None, release=False):
        """函数版-返回dict,代替request.values/request.json
        :param rules:参数的校验规则,map
        :param strip:对字段进行前后过滤空格
        :param default:将"" 装换成None
        :param diy_func:自定义的对某一参数的校验函数格式: {key:func},类似check, diy_func={"a": lambda x: x + "aa"})
        :param release:发生参数校验异常后是否依然让参数进入主流程函数
        """
        args_dict = OrderedDict()
        try:
            if request.values:
                args_dict.update(request.values)
            if request.json:
                args_dict.update(request.json)
            if release:
                args_dict_copy = copy.deepcopy(args_dict)  # 下面流程异常时,是否直接使用 原参数传入f # fixme
            # strip
            if strip:
                for k in args_dict:
                    if isstr(args_dict[k]):
                        args_dict[k] = args_dict[k].strip()
            # default
            if default[0]:
                for x in args_dict:
                    if args_dict[x] == "":
                        args_dict[x] = default[1]
            # diy_func
            if diy_func:
                for k in args_dict:
                    if k in diy_func:
                        args_dict[k] = diy_func[k](args_dict[k])
            # rules
            if rules:
                result, err = validate(rules, args_dict)
                if not result:
                    return False, err
        except Exception as e:
            print("verify_args catch err: ", traceback.format_exc())  # TODO
            if release:
                return True, args_dict_copy
            else:
                return False, str(e)
        return True, args_dict
    
    @app.route("/func", methods=["GET", "POST", "PUT"])
    def func_example():
        result, request_args = validator_func(rules=rules_example, strip=True)  # 姿势 2
        if not result:
            return jsonify({"code": 500, "data": None, "err": request_args})
        a = request_args.get("a")
        b = request_args.get("b")
        c = request_args.get("c")
        d = request_args.get("d")
        e = request_args.get("e")
        f = request_args.get("f")
        g = request_args.get("g")
        h = request_args.get("h")
        i = request_args.get("i")
        j = request_args.get("j")
        k = request_args.get("k")
        l = request_args.get("l")
        m = request_args.get("m")
        status, data = todo(a=a, b=b, c=c, d=d, e=e, f=f, g=g, h=h, i=i, j=j, k=k, l=l, m=m)
        if status:
            return jsonify({"code": 200, "data": data, "err": None})
        else:
            return jsonify({"code": 500, "data": None, "err": data})

嗯,还行吧,就是不怎么优雅,还是有点喜欢装饰器版本,但是苦于能力有限,不想看ImmutableMultiDict,MultiDict的实现,还是将第一个版本融合一下吧,装饰route不行,装饰todo还不行吗.于是有了版本4

### 版本4 ###

def validator_args(rules, strip=True, default=(False, None), diy_func=None, release=False):
        """针对普通函数的参数校验的装饰器
        :param rules:参数的校验规则,map
        :param strip:对字段进行前后过滤空格
        :param default:将"" 装换成None
        :param diy_func:自定义的对某一参数的校验函数格式: {key:func},类似check, diy_func={"a": lambda x: x + "aa"})
        :param release:发生参数校验异常后是否依然让参数进入主流程函数
        """
    
        def decorator(f):
            @wraps(f)
            def decorated_func(*args, **kwargs):
                if release:
                    args_bak = args[:]
                    kwargs_bak = copy.deepcopy(kwargs)  # 下面流程异常时,是否直接使用 原参数传入f # fixme
                try:
                    args_template = f.func_code.co_varnames
                except:
                    args_template = f.__code__.co_varnames
                args_dict = OrderedDict()
                try:
                    for i, x in enumerate(args):
                        args_dict[args_template[i]] = x
                    sorted_kwargs = sort_by_co_varnames(args_template, kwargs)
                    args_dict.update(sorted_kwargs)
                    # strip
                    if strip:
                        for k in args_dict:
                            if isstr(args_dict[k]):
                                args_dict[k] = args_dict[k].strip()
                    # default
                    if default[0]:
                        for x in args_dict:
                            if args_dict[x] == "":
                                args_dict[x] = default[1]
                    # diy_func
                    if diy_func:
                        for k in args_dict:
                            if k in diy_func:
                                args_dict[k] = diy_func[k](args_dict[k])
                    # rules
                    if rules:
                        result, err = validate(rules, args_dict)
                        if not result:
                            return False, err
                except Exception as e:
                    print("verify_args catch err: ", traceback.format_exc())
                    if release:
                        return f(*args_bak, **kwargs_bak)
                    else:
                        return False, str(e)
                return f(*args_dict.values())
    
            return decorated_func
    
        return decorator
        
        
    @validator_args(rules=rules_example, strip=True)  # 姿势 3
    def todo(a, b, c, d, e, f, g, h, i, j, k, l, m):
        return True, {"a": a, "b": b, "c": c, "d": d, "e": e, "f": f, "g": g, "h": h, "i": i, "j": j, "k": k, "l": l,
                      "m": m}

哎,就这样吧,打包一下,随便选吧,爱用哪个用哪个,反正我都写出来了.简单说就是:

1.  validator\_func 针对flask的request.json/requests.values的参数校验以及修改,修改的方式有限,可以自己控制
2.  validator\_wrap 是针对flask route的装饰器,针对request.json/requests.values的参数校验,只是校验,当然校验的方式可以自己写扩展
3.  validator\_args 针对普通函数的参数校验以及修改,注意不要使用python传参的高级特性(一个参数对应多个值),这个方法可以脱离flask使用,所以如果需要就直接copy过去吧.

嗯,最后还是分享一下到git上吧, [https://github.com/aleimu/flask-validator][https_github.com_aleimu_flask-validator] 喜欢的点个star.

[Python]: http://www.jqhtml.com/31186.html
[https_github.com_keleshev_s...]: https://github.com/keleshev/schema
[https_github.com_kvesteri_v...]: https://github.com/kvesteri/validators
[https_github.com_mansam_val...]: https://github.com/mansam/validator.py
[https_github.com_aleimu_flask-validator]: https://github.com/aleimu/flask-validator