Python webargs 模块的简单使用
时间：2022-05-16 08:46:05|栏目：Python代码|点击： 次
webargs是一个用于解析和验证HTTP请求对象的Python库，内置了对流行web框架的支持，包括Flask、Django、Bottle、Tornado、Pyramid、webapp2、Falcon和aiohttp。下面看下Python webargs 模块详解。
一、安装
python3 -m pip install webargs
二、基础特性
# encoding=utf-8
from flask import Flask
from webargs import fields
from webargs.flaskparser import use_args
app = Flask(__name__)
app.route("/")
@use_args({
"name": fields.Str(required=True),
"age": fields.Int(required=True),
}, location='query')
def index(args):
print('args', args)
return "Hello " + args["name"]
if __name__ == "__main__":
app.run(debug=1)
2.1. 使用
2.1.1 通过装饰器
@use_args({
"name": fields.Str(required=True),
"age": fields.Int(required=True),
}, location='query')
- 第一个参数是需要获取的字段名，类型，是否必须等的定义
- location是指从哪里获取这些参数，默认是json，可选：
- 'querystring' (same as 'query')
- 'json'
- 'form'
- 'headers'
- 'cookies'
- 'files'
解析完后，把所有参数放在字典里面，传给下层函数
2.1.2 通过函数
args = parser.parse(user_args, request)
参数和装饰器一样，多了一传request
2.2 参数检查
from webargs import fields, validate
args_1 = {
# 必须参数，字符串类型
"username": fields.Str(required=True),
# validate
"password": fields.Str(validate=lambda p: len(p) >= 6),
"password": fields.Str(validate=validate.Length(min=6)),
# Default value when argument is missing
"display_per_page": fields.Int(missing=10),
# Repeated parameter, e.g. "/?nickname=Fred&nickname=Freddie"
"nickname": fields.List(fields.Str()),
# Delimited list, e.g. "/?languages=python,javascript"
"languages": fields.DelimitedList(fields.Str()),
# When value is keyed on a variable-unsafe name
# or you want to rename a key
"user_type": fields.Str(data_key="user-type"),
"start_day": fields.DateTime(required=True, format='%Y-%m-%d %X'),
"bool": fields.Bool(),
"email": fields.Email(),
"ip": fields.IP(),
"type": fields.Constant(constant='COMMON_TYPE'),
"money": fields.Decimal(),
"weight": fields.Float(),
"url": fields.URL(),
"uuid": fields.UUID(),
"raw": fields.Raw(),
}
- fields.Str 表示接收字符串参数
- required=True 表示必传
- validate=lambda p: len(p) >= 6 表示自定义检查函数。会把参数传递给该函数，该函数返回True表示检查通过，返回False或者抛出异常表示检查不通过
- 如果要对多个参数进行联合检查，需要在装饰器层架validate参数：@use_args(args_1, validate=lambda args: len(args["username"]) < len(args["password"]))
- 异常需要是from webargs import ValidationError这个异常，不然会当程序异常处理
- 也可以用validate库里面的内置检查函数
- missing=10 表示如果没有入参，设置为默认值
- fields.List(fields.Str()) 表示列表型参数，列表的元素是字符串
- fields.DelimitedList(fields.Str()) 表示逗号型的列表参数
- data_key="user-type" 表示字段名修改，入参是user-type，在args字典会改为user_type
- fields.DateTime(required=True, format='%Y-%m-%d %X') 表示接收日期类型，并且格式需要符合，参数值会转换为datetime类型
- "bool": fields.Bool() 表示布尔类型，传1,0，true，false都能识别
- fields.Email() 只接收email，估计里面会有正则检查
- fields.IP() 只接收IP
- fields.Constant(constant='COMMON_TYPE') 常量参数，无论入参是什么值，type永远等于COMMON_TYPE
- fields.Decimal() 转换为Decimal类型
- fields.Float() 转换为float类型
- fields.URL() fields.UUID() 正则检查url格式或者uuid格式
- fields.Raw 不检查参数类型
内置参数检查
- validate=validate.Length(min=1,max=10) 检查字符串长度需要在某个区间
- validate=validate.OneOf(['male', 'female']) 入参需要在枚举里面
2.3 检查失败处理
如果参数检查失败，会返回422响应，但是不会提示哪个参数有问题。我们可以通过Flask的异常处理机制，捕获这个异常，然后构造我们想要的返回
@app.errorhandler(422) # 捕获422和400的异常码
@app.errorhandler(400)
def handle_error(err):
headers = err.data.get("headers", None)
messages = err.data.get("messages", ["Invalid request."])
print(headers)
print(messages) # {'json': {'password': ['Shorter than minimum length 6.']}}
return json.dumps({'err_code': 10000, 'err_msg': messages['json']})
- 从err里面获取信息，headers不知道有什么用的，message会有异常信息，例如不满足
validate=validate.Length(min=6)检查，就会返回{'json': {'password': ['Shorter than minimum length 6.']}} - 如果是主动抛出的ValidationError异常，message会包含ValidationError异常的内容
- 我们可以把这个参数检查信息返回给前端，提示前端哪个参数错误了。
messages['json']的json是location的key
2.4 嵌套参数
对于一些复杂的，有多重嵌套的参数
"name": fields.Nested(
{"first": fields.Str(required=True), "last": fields.Str(required=True)}
)
- 表示name是一个嵌套参数，也就是字典
- 然后里面需要要first key和last key
三、高级特性
3.1 自定义location
上面说了location支持query，json这些，也可以自定义
@parser.location_loader("data")
def load_data(request, schema):
data = {}
data.update({k: request.args.get(k) for k in request.args})
if request.json:
data.update({k: request.json.get(k) for k in request.json})
print(data, 'dataaaaa')
return data
parser.location = 'data' # 设置默认的location为data
- 上面定义了一个data的location，会合并args和json入参
- 把默认的location修改为data
也可以这样，这个是官方推荐方法：
@parser.location_loader("args_and_json")
def load_data(request, schema):
from webargs.multidictproxy import MultiDictProxy
newdata = request.args.copy()
if request.json:
newdata.update(request.json)
return MultiDictProxy(newdata, schema)
3.2 定义schema
除了可以通过字典定义args，也可以通过类：
from marshmallow import Schema
class UserSchema(Schema):
name = fields.Str(required=True)
age = fields.Int()
@app.route("/")
@use_args(UserSchema())
def index1(args):
print('args', args)
return "Hello "
3.3 未定义的参数处理
如果入参有未定义的参数，webargs默认会抛出422异常
from webargs.flaskparser import parser import marshmallow parser.unknown = marshmallow.EXCLUDE # 如果有未定义参数，不放到args参数，不抛异常 parser.unknown = marshmallow.INCLUDE # 如果有未定义参数，放到args参数，不抛异常
- 可以修改parse.unknown来修改策略。
- 也可以精确设置不同location的unknown策略
3.4 Flask的url参数
@app.route("/<int:id>/")
@use_args(UserSchema())
def index1(args, id):
print('args', args, id)
return "Hello "
如果需要用到Flask的url参数，就需要这样传递参数
