Python学习(flask-apispec简洁之道)

发表于 分类于 阅读次数： 本文字数： 4.8k 阅读时长 ≈ 4 分钟

在使用python写web框架时，经常会碰到需要对request参数进行检验或者过滤，如果将诸多的校验逻辑都堆积在业务逻辑中，会显得很臃肿，在flask中，推荐一个很棒的库，可以写法变得很清晰.

比如有这么个需求: 在flask中，有个app route需要验证requests的post中body传递过来的token字段必须在配置文件中才合法，同时phone字段需要符合规则，再同时，除了必要的字段外，不能有其它的字段，如果出现其它的字段，但返回指定的错误码.
如果在代码中直接写各种判断也是ok，之前作者也常是这么做，但明显不够优雅，引入flask-apispec就可以很好的解决问题，让代码看上去很简洁.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
#!/usr/local/bin/python 
#-* coding: utf-8-*-

import os 
import re 
import json 
import fire 
import requests
from config import cfg as CFG 
from loguru import logger
from utils.Dingtalk import DingtalkChatbot 
from pkg resources import require
from marshmallow import fields, ValidationError
from flask import request, Flask, jsonify, make_response from flask_apispec import use_kwargs
from tenacity import retry, stop_after_attempt, wait_fixed

app.Flask(__name__)
app.Config['JSON_AS_ASCII'] = False

# 处理出现422及400错误，返回对应的信息，发送回403
# Hand http error code 422 and 400 
@app.Errorhandler(422) 
@app.Errorhandler(400) 
def handle_error(err):
  msg = err.Data.Get("messages",  ["Invalid Request"]) 
  logger.Critical(msg)
  return make_response(jsonify({"msg": msg["json"]}), 403)

# 这是dingding发送逻辑，不重要
def dingtalk(**kw):
  is_at_all=False
  content = kw.Get("content", "alarm: THIS-IS-DEFAULT-CONTENT-FROM-DINGTALK") 
  token = kw.Get("token") 
  to = kw.Get("to") 
  if not token or not to:
    logger.Error("token or to can't be null... ")
  else:
    if "@all" in to:
      is_at_all = True
      tos = []
    else:
      tos = [x for x in to.Replace (" ", "").Split(", ")]
      logger.Info ("alert [dingtalk]: {}".format(content))
      xiaoding.DingtalkChatbot(**{"token": token})
      xiaoding.Send_text(msg="alarm: " + content, at_mobiles=tos, is_at_all=is_at_all)

# 正则判断手机号是否合法
def phone_check(phone):
  phone_rule="^((13[0-9])|(14[5,7])|(15[0-3,5-9])|(17[0,3,5-8])|(18[0-9])166|198|199|(147))\\d{8}$"
  return True if re.Match(phone_rule, phone.Strip()) else False

# 检验发送者手机号是否合法
def check_sender(to):
  if "@all" in to:
    return True
  else:
    for x in to.Strip().Split (", "):
      if not x.Strip().Isdigit() or not phone_check(x):
        raise ValidationError("phone NotFound or number is invalid")

# 查看token是否在配置文件中, 配置文件是个yaml格式，格式不重要.
def must_exist_in_conf(t):
  if t.Strip() not in CFG["dingtalk"]["tokens"]:
    raise ValidationError("token is invalid")

# 重要
args_dt = {"token": fields.Str(required=True, validate=must_exist_in_conf), "content": fields.Str(required=True), "to": fields.Str(load_default="@all", validate=check_sender)}

# 重要
# use kwargs dict field will pass to **kw args
@app.Route("/webhook/dingtalk", methods=["POST"]) 
@use_kwargs(args_dt, location="json") 
def webhook_dingtalk(**kw):
  _dingtalk (**kw)
  return make_response(jsonify({"msg": "OK"}), 200)

# 启动flask
def runapp():
  app.Run(host='0.0.0.0', port=5555)

def watchdog():
  pass

if name == " main ":
  fire. Fire({"runapp": runapp, "watchdog": watchdog})

其它的也没什么，重要的只的@use_kwargs装饰器,接下来详细展开

1
2
3
4
5
@app.Route("/webhook/dingtalk", methods= ["POST"]) 
@use_kwargs(args_dt, location="json") 
def webhook_dingtalk(**kw):
  _dingtalk (**kw)
  return make_response(jsonify({"msg": "OK"}), 200)

如果按照正常的写法一般是

1
2
3
4
5
6
@app.Route("/webhook/dingtalk", methods=["POST"]) 
def webhook_dingtalk():
  res = request.json()
  # 调用各种逻辑对参数进行判断
  _dingtalk (**res)
  return make_response(jsonify({"msg": "OK"}), 200)

在函数中直接使用request.json()获取body的key-value, 然后做各种判断
那现在主函数webhook_dingtalk(**kw)非常简短，答案就在引入了use_kwargs,简单来讲就是use_kwargs将接收request的参数，然后将参数作用于第一个参数指定的字典，即args_dt，简单看一下args_dt

1
args_dt = {"token": fields.Str(required=True, validate=must_exist_in_conf), "content": fields.Str(required=True), "to": fields.Str(load_default="@all", validate=check_sender)}

其实也是非常清晰，在这个字典中其它指定了参数列表，也指定了各个参数需要进行的判断,比如token字段，required=True,表明这个字段是必要的，validate则表示这个字段需要进行的逻辑处理, 那么这里就可以写各种业务上的判断了,其实内置了很多常用的一些规则，比如判断是不是布尔型fields.Boolean()等等.
这种写法是不是让主函数看上去简洁了许多.
主函数webhook_dingtalk(**kw)接收的参数只有kw, kw其它就是args_dt合法后传递过来，如果从requst中拿到的参数不符合args_dt中指定的任一规则，则会触发422或者400错误，将会调用代码最上面定义的handle_error自定义逻辑，这里是返回403错误

1
2
def use_kwargs(args, locations=None, inherit=None, apply=None, **kwargs):
    pass

另外, use_kwargs的第2个参数location,表示的是从http传递过来的参数是以什么方式呈现，可以选用(‘json’, ‘querystring’, ‘form’, ‘headers’, ‘cookies’, ‘files’)作为locations的值,因为参数可以放在body中，也可以放在header或者是cookies中，这个参数主要是告诉use_kwargs需要的参数保存在哪里.
flask-apispec内部用了webargs用于参数解析, marshmallow库用于返回响应,也用了apispec，还有一些很实用的功能，感兴趣的可以查看官网
使用flask-apispec, 代码看上去舒服多了

参考文章: