给每个flask接口增加验证调用装饰器

Posted lishanlu136

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了给每个flask接口增加验证调用装饰器相关的知识,希望对你有一定的参考价值。

碰到这个问题是在做服务端算法部署的时候碰到的,我需要将算法服务做成flask接口以供平台后端调用,但启动算法需要先通过服务器硬件验证,验证通过方能正常启动。为防止避开验证功能启动算法。就需要在验证功能函数中加一个verify_flag,当验证通过过,修改verify_flag为True,当调用接口时,首先验证verify_flag是否为True,当为True时,才能正常调用接口,否则,停止调用。

1、原本的flask接口,service.py

import sys
from flask import Flask,request
import json
import func2_test
import func4_test
import func5_test

app = Flask(__name__)

@app.route("/test_1", methods=['POST'])
def test_1():
    data = request.form
    try:
        x = float(data.get('x', 1.0))
        y = float(data.get('y', 2.0))
        z = func2_test(x, y)
        m = func4_test(x, y, z)
        result = 'status': 0, 'msg': '', 'result': m
    except Exception as e:
        result = 'status': 1, 'msg': str(e), 'result': -1
        
    return json.dumps(result)


@app.route("/test_2", methods=['POST'])
def test_2():
    data = request.form
    try:
        x = float(data.get('x', 1.0))
        y = func5_test(x)
        m = func2_test(x, y)
        result = 'status': 0, 'msg': '', 'result': m
    except Exception as e:
        result = 'status': 1, 'msg': str(e), 'result': -1
        
    return json.dumps(result)


@app.route("/hello", methods=['GET'])
def hello():
    return "hello world" 

if __name__ == '__main__':
	app.run(host='0.0.0.0', port=8202)

2、加入装饰器@verify

2.1 在read_license.py文件中设置verify_flag值作为验证标志,初始值为False,通过里面的verify()函数修改为True.

verify_flag = False

def verify():
    if '3' == '3':         # 此处相当于验证条件,当条件满足时,即验证通过,修改verify_flag为True
        global verify_flag
        verify_flag = True
        return 1
    else:
        return 0

2.2 在service.py中导入read_license.py中的verify_flag,通过这个值来判断是否正常调用接口

import sys
from flask import Flask,request
import json
import func2_test
import func4_test
import func5_test
import read_license as rl
import functools

app = Flask(__name__)

def verify(func):
    @functools.wraps(func)                  # 必须加,不然会报错
    def is_verify():
        if rl.verify_flag:
            print("verify flag:", rl.verify_flag)
            return func()
        else:
            print("License verify fail.")
            print("verify flag:", rl.verify_flag)
            sys.exit()
    return is_verify


@app.route("/test_1", methods=['POST'])
@verify
def test_1():
    data = request.form
    try:
        x = float(data.get('x', 1.0))
        y = float(data.get('y', 2.0))
        z = func2_test(x, y)
        m = func4_test(x, y, z)
        result = 'status': 0, 'msg': '', 'result': m
    except Exception as e:
        result = 'status': 1, 'msg': str(e), 'result': -1

    return json.dumps(result)


@app.route("/test_2", methods=['POST'])
@verify
def test_2():
    data = request.form
    try:
        x = float(data.get('x', 1.0))
        y = func5_test(x)
        m = func2_test(x, y)
        result = 'status': 0, 'msg': '', 'result': m
    except Exception as e:
        result = 'status': 1, 'msg': str(e), 'result': -1

    return json.dumps(result)


@app.route("/hello", methods=['GET'])
@verify
def hello():
    return "hello world"

if __name__ == '__main__':
    if rl.verify():            # 验证,如果验证通过,修改verify_flag为True
        print("verify flag:", rl.verify_flag)
        app.run(host='0.0.0.0', port=8202)
    else:
        print("verify fail.")
        sys.exit()

2.3 容易出问题的地方

如果装饰器函数这样写

def verify(func):
    def is_verify():      # 注意,没加@functools.wraps(func)
        if rl.verify_flag:
            print("verify flag:", rl.verify_flag)
            return func()
        else:
            print("License verify fail.")
            print("verify flag:", rl.verify_flag)
            sys.exit()
    return is_verify

@app.route("/test_1", methods=['POST'])
@verify
def test_1():
    ...
    return json.dumps(result)


@app.route("/test_2", methods=['POST'])
@verify
def test_2():
    ...
    return json.dumps(result)


@app.route("/hello", methods=['GET'])
@verify
def hello():
    return "hello world"

启动服务的时候就会报错:

仔细想想就明白为什么会报错了,因为函数经过verify装饰后,返回的函数名字是is_verify了,不再是装饰之前的函数名了,所以修改方式有两种,一种就是之前的方式,在装饰函数中增加@functools.wraps(func),另外一种就在@app.route()增加endpoint参数,指定函数名:

def verify(func):
    def is_verify():      # 注意,没加@functools.wraps(func)
        if rl.verify_flag:
            print("verify flag:", rl.verify_flag)
            return func()
        else:
            print("License verify fail.")
            print("verify flag:", rl.verify_flag)
            sys.exit()
    return is_verify

@app.route("/test_1", methods=['POST'], endpoint='test_1')
@verify
def test_1():
    ...
    return json.dumps(result)


@app.route("/test_2", methods=['POST'], endpoint='test_2')
@verify
def test_2():
    ...
    return json.dumps(result)


@app.route("/hello", methods=['GET'], endpoint='hello')
@verify
def hello():
    return "hello world"

以上是关于给每个flask接口增加验证调用装饰器的主要内容,如果未能解决你的问题,请参考以下文章

给每个flask接口增加验证调用装饰器

Flask 给视图函数增加装饰器

为 Flask 创建身份验证装饰器的问题

用装饰器封装Flask-WTF表单验证逻辑

Flask--登录验证(多个装饰器)

Python Flask装饰器登录验证