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

Posted 2023-02-24 lishanlu136

碰到这个问题是在做服务端算法部署的时候碰到的，我需要将算法服务做成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"