测试平台系列(34) 编写全局变量接口

回顾

上一节咱们编写好了环境管理功能，这一章节我们来继续完善全局变量功能。

全局变量?

全局变量，其实我觉得叫它全局配置更加贴切。我理解的全局变量，其实是我们常用的一些不太变化的数据，而用例中出现的变量，我认为它是一个临时的数据，不会进行永久存储。

那什么时候会用到这些变量呢？比如咱们的一些常用的测试地址，ip也好url也好，如果用例不把这块内容抽离出来，一旦地址发生变化，会出现2个问题。

我们需要对用例进行修改
修改的成本较大

有同学可能会说，那我去数据库里面直接update一下不就好了？是的，你可以这么做。但是谁能保证update的时候不会出错，谁又能保证每一次都能及时update了呢？所以，如果我能在全局变量页面进行统一修改，那不是事半功倍吗？这也就是我所说的，它其实更像一个全局配置。

设计思路

其实我们提供的全局配置，类似于Redis，是一个key-value的形式。用户如果使用的话，只需要输入key，就能够映射到对应的value。

想象一下，如果一个value能够存放int，str, float等多种对象，那么Mysql里面这个字段该如何设计？所以这个时候我们就要加以辅佐，我们引入一个key_type字段，配合value实现一个字段多种数据类型的功能。

但我这边又不想太复杂，因为咱都是懒人。其实我们定义几种类型就够了:

string
json
yaml

为什么只有3种类型？因为已经够了，如果是string类型，则不需要进行json.loads()去获取数据，如果是json，我们默认通过json.loads()去转换一下数据。至于yaml，纯粹是为了新鲜。

为什么说我们用json就可以拿到float,int,bool等等值呢？我们来看一个例子:

证明一下

那么我们存在数据库的值是varchar类型的0.618，那我们放入Python可以表现为这样:

a = "0.618"

那怎么得到float呢？

a = "0.618"
import json
a = json.loads(a)
print(type(a), a)

image.png

同理试一下bool:

image.png

至于string为什么和json区分开，其实json就是个string，所以我们如果遇到string类型就不做任何json.loads的处理，即可保留原滋原味。

设计表

image.png

env: 变量对应的环境，如果env为0的话，说明这个变量属于全部环境。
enable: 变量是否可用，是一个开关字段，可临时关闭该变量

注意这里有一个联合唯一索引，根据env,key和deleted_at生成，只有当env,key,deleted_at都相等的时候，数据库才不让插入数据。这样解决了一个在environment表里面出现的问题:

当环境abc被删除以后，无法再建立名为abc的环境

image.png

注册model

image.png

编写schema

[图片上传失败...(image-423036-1624776054767)]

编写crud功能

from datetime import datetime

from sqlalchemy import desc

from app.models import Session, update_model
from app.models.gconfig import GConfig
from app.models.schema.gconfig import GConfigForm
from app.utils.logger import Log


class GConfigDao(object):
    log = Log("GConfigDao")

    @staticmethod
    def insert_gconfig(data: GConfigForm, user):
        try:
            with Session() as session:
                query = session.query(GConfig).filter_by(env=data.env, key=data.key, deleted_at=None).first()
                if query is not None:
                    return f"变量: {data.key}已存在"
                config = GConfig(**data.dict(), user=user)
                session.add(config)
                session.commit()
        except Exception as e:
            GConfigDao.log.error(f"新增变量: {data.key}失败, {e}")
            return f"新增变量: {data.key}失败, {str(e)}"
        return None

    @staticmethod
    def update_gconfig(data: GConfigForm, user):
        try:
            with Session() as session:
                query = session.query(GConfig).filter_by(id=data.id, deleted_at=None).first()
                if query is None:
                    return f"变量{data.key}不存在"
                update_model(query, data, user)
                session.commit()
        except Exception as e:
            GConfigDao.log.error(f"编辑变量失败: {str(e)}")
            return f"编辑变量失败: {str(e)}"
        return None

    @staticmethod
    def list_gconfig(page, size, env=None, key=None):
        try:
            search = [GConfig.deleted_at == None]
            with Session() as session:
                if env:
                    search.append(GConfig.env == env)
                if key:
                    search.append(GConfig.name.ilike("%{}%".format(key)))
                data = session.query(GConfig).filter(*search)
                total = data.count()
                return data.order_by(desc(GConfig.created_at)).offset((page - 1) * size).limit(
                    size).all(), total, None
        except Exception as e:
            GConfigDao.log.error(f"获取变量列表失败, {str(e)}")
            return [], 0, f"获取变量列表失败, {str(e)}"

    @staticmethod
    def delete_gconfig(id, user):
        try:
            with Session() as session:
                query = session.query(GConfig).filter_by(id=id).first()
                if query is None:
                    return f"变量{id}不存在"
                query.deleted_at = datetime.now()
                query.update_user = user
                session.commit()
        except Exception as e:
            GConfigDao.log.error(f"删除变量失败: {str(e)}")
            return f"删除变量失败: {str(e)}"
        return None

看过上一篇文章的都知道细节，其实我写代码的时候也是copy的。

大家有没有发现: 编写一个crud的功能是真的不难！基本可以按照固定的套路来，像我都是copy的上一个功能点的代码，然后稍作改动。

编写核心接口

from fastapi import Depends

from app.dao.config.GConfigDao import GConfigDao
from app.handler.fatcory import ResponseFactory
from app.models.schema.gconfig import GConfigForm
from app.routers import Permission
from app.routers.config.environment import router
from config import Config


@router.get("/gconfig/list")
async def list_gconfig(page: int = 1, size: int = 8, env: int = None, key: str = "", user_info=Depends(Permission())):
    data, total, err = GConfigDao.list_gconfig(page, size, env, key)
    if err:
        return dict(code=110, msg=err)
    return dict(code=0, data=ResponseFactory.model_to_list(data), total=total, msg="操作成功")


@router.post("/gconfig/insert")
async def insert_gconfig(data: GConfigForm, user_info=Depends(Permission(Config.ADMIN))):
    err = GConfigDao.insert_gconfig(data, user_info['id'])
    if err:
        return dict(code=110, msg=err)
    return dict(code=0, msg="操作成功")


@router.post("/gconfig/update")
async def update_gconfig(data: GConfigForm, user_info=Depends(Permission(Config.ADMIN))):
    err = GConfigDao.update_gconfig(data, user_info['id'])
    if err:
        return dict(code=110, msg=err)
    return dict(code=0, msg="操作成功")


@router.get("/gconfig/delete")
async def delete_gconfig(id: int, user_info=Depends(Permission(Config.ADMIN))):
    err = GConfigDao.delete_gconfig(id, user_info['id'])
    if err:
        return dict(code=110, msg=err)
    return dict(code=0, msg="操作成功")

其实大体上还是和之前的router差不多，但是这里多了一个细节: