用PyWebio编写一个简易造数平台-六虎

我正在参加「掘金启航计划」

布景

如果是经过Django或Flask完结WEB运用的话，写起来比较麻烦。偶然发现一个Python第三方库—PyWebIO，能够用来快速构建WEB运用。所以，用它来测验写一个简易的造数据渠道，用于批量造一些事务数据，辅助测验。

一、简介

1.什么是pywebio

PyWebIO是一个能够用来快速构建WEB运用的开源的Python第三方库。以下内容来自官方简介：

PyWebIO供给了一系列指令式的交互函数来在浏览器上获取用户输入和进行输出，将浏览器变成了一个“富文本终端”，能够用于构建简略的Web运用或根据浏览器的GUI运用。运用PyWebIO，开发者能像编写终端脚本相同(根据input和print进行交互)来编写运用，无需具有HTML和JS的相关知识； PyWebIO还能够方便地整合进现有的Web服务。非常适合快速构建对UI要求不高的运用。

官方教程：pywebio.readthedocs.io/zh_CN/lates…

项目地址：github.com/pywebio/PyW…

2.特性

运用同步而不是根据回调的办法获取输入，代码编写逻辑更天然
非声明式布局，布局办法简略高效
代码侵入性小，旧脚本代码仅需修正输入输出逻辑便可改造为Web服务
支持整合到现有的Web服务，目前支持与Flask、Django、Tornado、aiohttp、 FastAPI(Starlette)结构集成
同时支持根据线程的履行模型和根据协程的履行模型
支持结合第三方库完结数据可视化

二、装置及用法示例

1.装置

pip3 install -U pywebio  # 装置最新版
pip3 install pywebio==1.4.0  # 装置指定版别

体系要求: PyWebIO要求 Python 版别在 3.5.2 及以上

2.示例

以下示例来自官方教程，这是一个运用PyWebIO计算 BMI指数的脚本：

from pywebio.input import input, FLOAT
from pywebio.output import put_text
def bmi():
    height = input("请输入你的身高(cm)：", type=FLOAT)
    weight = input("请输入你的体重(kg)：", type=FLOAT)
    BMI = weight / (height / 100) ** 2
    top_status = [(14.9, '极瘦'), (18.4, '偏瘦'),
                  (22.9, '正常'), (27.5, '过重'),
                  (40.0, '肥壮'), (float('inf'), '非常肥壮')]
    for top, status in top_status:
        if BMI <= top:
            put_text('你的 BMI 值: %.1f，身体状况：%s' % (BMI, status))
            break
if __name__ == '__main__':
    bmi()

作用如下：

三、造数渠道实战

1.适用场景（目的）

例如：我想测验一个前台页面数据分页是否正常，一个分页默认最大20条数据，需求造20+条数据。常见的造数据办法如下：

手工一条条增加：效率太低、数据量大的情况下不引荐，也不现实；
编写SQL语句：履行SQL进行批量增加，前提是需求弄清楚表结构信息，掌握SQL语法；
恳求接口来增加：这也是测验进程中最常用的办法，效率高、无需了解表结构，缺点是依靠较多、链路较长的时候比较难处理；
造数据渠道：运用简略，无需清楚接口调用、表结构，但其背面仍然是恳求接口或直接履行SQL语句；

可见，数据渠道的办法能够便于不会编码的同学或其他项目组同事快速批量结构事务数据，辅助测验或体系演示。

2.规划思路

整个前台页面操作流程如下：

用户输入账号暗码、登录体系，留意这里的体系不是造数据体系，而是要结构数据的事务体系；
挑选环境，例如：开发环境、测验环境、预发布环境等；
挑选数据类别，例如：乒乓球、篮球、羽毛球；
填写要结构数据的数量，能够是多条；
提交，等候结构结果。

其背面的完结逻辑如下：

用户登录，恳求登录接口，获取到用户信息、企业信息等，供后续接口传参运用；
挑选环境，挑选不同的环境就会读取不同的装备；
挑选数据类型，获取到数据类型参数，作为判别条件，以决议恳求哪个接口；
填写数量，拿到数值，作为for循环次数，决议结构多少条数据；
最后提交，将各个参数传递给造数据接口，恳求接口（接口结构数据背面仍履行的是SQL）；

3.用到的pywebio相关的办法

设置网页标题

pywebio.session.set_env(title="RS造数渠道")

输入文本

input.input(label="请输入登录手机号：")

设置图片

img = open('E:/RS_OTMS.png', 'rb').read()
output.put_image(img)

设置挑选框

select = input.select(label="请挑选环境", options=["测验环境", "开发环境"])

输出进展条、指定文本

进展条是以百分比的方式展现，0.4即40%

# 输出进展条为40%
output.set_processbar("当时进展", 0.4)
# 输出指定文本内容
output.put_text(f"登录成功，登录手机号为: {phone_number}，当时企业类型为: 车队")

发动服务

applications：要运行的函数名，留意不要带()；
host：设置的本地地址，不填的话默认0.0.0.0；
port：本地端口号；

pywebio.start_server(applications=run, host='192.168.1.131', port=8888)

4.编码完结

导入相应库

faker库用于生成数据，如姓名、手机号、身份证号码、银行卡号、地址等等；
pywebio的input用于用户输入，output用于输出展现自界说内容到前台；
MySQL是自界说的一个类，用于结构数据进程中和接口进行合作验证等运用；
config是一个自界说的用于寄存装备的装备文件；
AuvDataProvider、RsCommon、RSResource、RSPool是事务的接口类；

from faker import Faker
import hashlib
import pywebio
from pywebio import input, output
import config
from execute_mysql import MySQL
from auv_data_provider import AuvDataProvider
from rs_common import RsCommon
from rs_resource import RSResource
from rs_pool import RSPool

界说获取输入信息的办法

例如：获取手机号、暗码，用于登录，然后获取到用户信息，供后续接口传参运用。留意：

因为事务体系中有多个类型的企业，登录人归于某个企业下，所以为了完结能够给不同企业增加数据，此处的登录账号不宜写死；
每个账号的暗码也不一定相同，所以此处也不宜写死；
因为前台输入的暗码是一般明文字符，但在后台传参时用的是MD5加密后的密文，所以此处需求界说一个函数将暗码转换为MD5格局，参数是前台获取到的手动输入的暗码；

def get_phone_number():
    """获取手机号"""
    phone_number = input.input(label="请输入登录手机号：")
    return phone_number
def get_pwd():
    """获取登录暗码"""
    pwd = input.input(label="请输入登录暗码")
    return pwd
def get_numbers():
    """获取增加数据的数量"""
    numbers = input.input(label="请填写数量：")
    return int(numbers)
def transform_pwd_to_md5(pwd: str):
    """将暗码转换为MD5格局"""
    m = hashlib.md5(pwd.encode(encoding='utf-8'))
    pwd_md5 = m.hexdigest()
    return pwd_md5

界说增加数据的主函数

实际上便是将上述pywebio常用的办法同各个接口调用、依照上述的规划思路相结合。此处主要以快速完结功能为主，暂不考虑什么功能、代码精简度等，完好代码如下：

def run():
    """增加数据主函数"""
    # 声明全局变量
    global url, mysql, cpy_id, cpy_user_name, cpy_name, user_id
    # 自界说布景图
    img = open('E:/RS_OTMS.png', 'rb').read()
    output.put_image(img)
    # 初始化数据库等
    product_id = "2022"
    select = input.select(label="请挑选环境", options=["测验环境", "开发环境"])
    if select == "开发环境":
        # 读取装备文件config的内容，不同环境读取不同装备
        url = config.URLConf.RS_DEV_URL.value
        mysql = MySQL(host=config.DBConfig.db_dev['host'], pwd=config.DBConfig.db_dev['pwd'],
                      ssh_host=config.DBConfig.db_dev['ssh_host'], ssh_pwd=config.DBConfig.db_dev['ssh_pwd'],
                      dbname=config.DBConfig.db_dev['dbname_rs'])
    elif select == "测验环境":
        url = config.URLConf.RS_TEST_URL.value
        mysql = MySQL(host=config.DBConfig.db_test['host'], pwd=config.DBConfig.db_test['pwd'],
                      ssh_host=config.DBConfig.db_test['ssh_host'], ssh_pwd=config.DBConfig.db_test['ssh_pwd'],
                      dbname=config.DBConfig.db_test['dbname_rs'])
    adp = AuvDataProvider()
    fake = Faker("zh_CN")
    common = RsCommon(product_id=product_id, url=url)
    resource = RSResource(product_id=product_id, url=url)
    pool = RSPool(product_id=product_id, url=url)
    # 设置网页标题
    pywebio.session.set_env(title="RS造数渠道")
    # 输入手机号
    output.put_processbar("当时进展")
    # 输入暗码
    phone_number = get_phone_number()
    output.set_processbar("当时进展", 0.2)
    # 暗码转换为MD5
    pwd = transform_pwd_to_md5(get_pwd())
    output.set_processbar("当时进展", 0.4)
    # 事务登录
    login = common.login(username=phone_number, password=pwd)
    user_id = login["b"]
    user_info = common.get_user_info(user_id=user_id)
    cpy_id = user_info["d"]["b"]  # b节点为企业ID,h节点为企业全称,j节点为登录人手机号
    cpy_type = user_info["d"]["l"]  # 企业类型: 1-旭阳, 2-客商, 3-车队
    output.put_text(login)
    # 企业类型为1的企业增加数据
    if cpy_type == 1:
        output.put_text(f"登录成功，登录手机号为: {phone_number}，当时企业类型为: 旭阳")
        data_type = input.select("请挑选增加数据的类型", ["公共池车辆", "公共池挂车", "公共池司机", "公共池押运员"])
        output.set_processbar("当时进展", 0.6)
        number = input.input("请输入要增加数据的数量: ")
        output.set_processbar("当时进展", 0.8)
        # 增加公共池运力数据逻辑
        if data_type == "公共池车辆":
            for i in range(int(number)):
                car_name = adp.create_car()
                add_car = pool.add_pool_car(cpy_id, user_id, car_name)
                output.put_text("增加车辆成功，车辆id为%s" % (add_car["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "公共池挂车":
            for i in range(int(number)):
                gua_name = adp.create_gua_car()
                add_gua_car = pool.add_pool_gua_car(cpy_id, user_id, gua_name)
                output.put_text("增加挂车成功，挂车id为%s" % (add_gua_car["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "公共池司机":
            for i in range(int(number)):
                add_driver = pool.add_pool_driver(cpy_id, user_id)
                output.put_text("增加司机成功，司机id为%s" % (add_driver["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "公共池押运员":
            for i in range(int(number)):
                escort_driver = pool.add_pool_escort(cpy_id, user_id)
                assert escort_driver["a"] == 200
                escort_user_id = escort_driver['d']
                output.put_text("增加押运员成功，押运员id为%s" % (escort_user_id))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
    # 企业类型为2的企业增加数据
    elif cpy_type == 2:
        output.put_text(f"登录成功，登录手机号为: {phone_number}，当时企业类型为: 车队")
        data_type = input.select("请挑选增加数据的类型", ["车辆", "挂车", "司机", "押运员", "运力"])
        output.set_processbar("当时进展", 0.6)
        number = input.input("请输入要增加数据的数量: ")
        output.set_processbar("当时进展", 0.8)
        # 增加车队运力数据逻辑
        if data_type == "车辆":
            for i in range(int(number)):
                car_name = adp.create_car()
                add_car = resource.add_car_simple_info(cpy_id, user_id, car_name, car_type=1)
                output.put_text("增加车辆成功，车辆id为%s" % (add_car["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "挂车":
            for i in range(int(number)):
                gua_name = adp.create_gua_car()
                add_gua_car = resource.add_gua_simple_info(cpy_id=cpy_id, user_id=user_id, gua_name=gua_name,
                                                           gua_type=10, origin=1)
                output.put_text("增加挂车成功，挂车id为%s" % (add_gua_car["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "司机":
            for i in range(int(number)):
                driver_name = fake.name()
                driver_phone = fake.phone_number()
                driver_id_card = fake.ssn()
                add_driver = resource.add_driver(user_id=user_id, cpy_id=cpy_id, driver_name=driver_name,
                                                 driver_phone=driver_phone, driver_id_card=driver_id_card,
                                                 driver_type=1)
                output.put_text("增加司机成功，司机id为%s" % (add_driver["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "押运员":
            for i in range(int(number)):
                escort_name = fake.name()
                escort_phone = fake.phone_number()
                escort_id_card = fake.ssn()
                escort_driver = resource.add_escort(user_id=user_id, cpy_id=cpy_id, escort_name=escort_name,
                                                    escort_phone=escort_phone, escort_id_card=escort_id_card,
                                                    escort_type=1)
                assert escort_driver["a"] == 200
                escort_user_id = escort_driver['d']
                output.put_text("增加押运员成功，押运员id为%s" % (escort_user_id))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
        elif data_type == "运力":
            for i in range(int(number)):
                # 新增车辆
                car_name = adp.create_car()
                add_car_simple = resource.add_car_simple_info(cpy_id, user_id, car_name, car_type=1)
                assert add_car_simple["a"] == 200
                car_id = add_car_simple['d']
                # 将车辆状况置为审阅经过
                mysql.execute_sql(f"UPDATE t_r_car_info SET review_status=1 WHERE id={car_id}")
                # 新增司机
                driver_name = fake.name()
                driver_phone = fake.phone_number()
                driver_id_card = fake.ssn()
                add_driver = resource.add_driver(user_id=user_id, cpy_id=cpy_id, driver_name=driver_name,
                                                 driver_phone=driver_phone, driver_id_card=driver_id_card,
                                                 driver_type=1)
                assert add_driver["a"] == 200
                driver_user_id = add_driver['d']
                # 将司机状况置为审阅经过
                mysql.execute_sql(
                    f"UPDATE t_r_driver_info SET review_status=1 WHERE user_id={driver_user_id}")
                driver_id = mysql.execute_sql(f"SELECT id FROM t_r_driver_info WHERE user_id={driver_user_id}")[
                    0]  # 查询司机user_id
                # 新增运力绑定
                add_trans = resource.add_transporter(cpy_id=cpy_id, user_id=user_id, car_id=car_id, driver_id=driver_id,
                                                     trans_name=car_name + driver_name + str(driver_phone))
                assert add_trans["a"] == 200
                output.put_text("新增运力成功，运力id为%s" % (add_trans["d"]))
            output.put_text(f"新增完结，本次共新增{number}条数据！")
    output.set_processbar("当时进展", 1)
if __name__ == '__main__':
    pywebio.start_server(applications=run, host='192.168.1.131', port=8888)

5.运用作用

整体作用如下：

小结

以上便是使用Python+pywebio库完结的一个简易数据结构渠道的整体进程，Python语法是根底，再学习一些pywebio常用的办法即可。本次主要是为了快速完结渠道并能够增加常用数据，并未过多关注规划形式、代码功能等等，也存在很多能够优化的当地！欢迎批评指正！

用PyWebio编写一个简易造数平台

布景