米洛的小窝

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

2022/6/13 测试平台接口测试FastApiPythonReact

# 回顾

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

# 全局变量?

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

那什么时候会用到这些变量呢？比如咱们的一些常用的测试地址，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"
1

那怎么得到float呢？

a = "0.618"
import json
a = json.loads(a)
print(type(a), a)
1
2
3
4

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.png

# 编写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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

看过上一篇文章的都知道细节，其实我写代码的时候也是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="操作成功")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

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

  • 当config router下面有个文件的时候会怎么样？

image.png

我们这里有一个gconfig和一个envrionment，里面大概8个接口。

envrionment定义了ApiRouter的router对象，gconfig引入了router对象。那么我们的接口会被成功注册吗？

答案是不会的，大家可以试一试。

因为顺序是这样的:

  1. router = ApiRouter() # environment中发生
  2. pity.include_router(config.router)
  3. router注册gconfig的接口

因为咱们运行的是main.py文件，main.py在import router的时候肯定是最先发生的。其实flask也会遇到这样的问题，解决方法有2个。

  • 快速法

    main.py注册router的时候，选用最后一个。

    举个例子，envrionment创建了router，gconfig引入了router，我们从gconfig import router就行。但这样不会太保险，因为还是可能会遗漏或者搞错。

  • 利用__init__.py

    image.png

    最后在注册的时候import init.py里面的router即可。简单的说就是搞一个专门收集router的文件，最后集中注册。

image.png

# 总结

本期除了写了全局变量功能以外，还解决了2个问题:

  • 联合索引问题
  • router注册问题

希望对大家有帮助~

后端地址: https://github.com/wuranxu/pity (opens new window)

前端地址: https://github.com/wuranxu/pity (opens new window)

关注我的个人公众号米洛的测开日记，催更不迷路。

(35) 编写全局变量管理页面