flask项目-请求测试

app.py文件--------创建项目、设置路由、启动项目

from flask import Flask
from flask_restful import Api
from concurrent.futures import ThreadPoolExecutor


executor = ThreadPoolExecutor(max_workers=10)


app = Flask(__name__)
app.executor = executor
api = Api(app)

from app.num_check import num_check
#添加路由接口
api.add_resource(num_check, "/num_check/LV/check_test/")

if __name__ == '__main__':
    app.run()

num_check.py------利用Resource模板设置get请求的响应信息

from flask_restful import Resource
from flask import request
from flask import current_app
import time

class num_check(Resource):
    def __init__(self):
        self.GET = request.args.to_dict()

    def get(self):
        num = request.args.to_dict().get("num")
        num = int(num)
        time.sleep(3)
        t = current_app.executor.submit(num_check_load, num)
        check_info = t.result()
        if check_info[0]:
            return {"code": 0, "info": "", "data": check_info[1]}
        else:
            return {"code": 1, "info": check_info[1], "data": ""}


def num_check_load(num):
    if num > 10:
        return [False, "%s 不能大于10" % num]
    check_info = "%s 通过" % num
    return [True, check_info]

check.py-------测试文件

import requests
from threading import Thread


def requests_result(URL):
    ret = requests.get(URL)
    if ret.status_code != 200:
        print("请求失败")

    ret_dect = ret.json()
    if ret_dect["code"] == 1:
        error_info = ret_dect["info"]
        print(error_info)

    else:
        print(ret_dect["data"])

    return ret


threading_fool = []
for i in range(100):
    URL = "http://127.0.0.1:5000//num_check/LV/check_test/?num=%s" % i
    t = Thread(target=requests_result, args=(URL,))
    threading_fool.append(t)

for i in threading_fool:
    i.start()

for i in threading_fool:
    i.join()

 

上一篇:docker 报错Failed to start Docker Storage Setup. 的处理基本都是容器满了


下一篇:可编译运行:调用ffmpeg接口,将RTSP流保存为MP4的C代码