zoukankan      html  css  js  c++  java
  • 测试平台系列(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.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
    
    

    看过上一篇文章的都知道细节,其实我写代码的时候也是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差不多,但是这里多了一个细节:

    • 当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

    前端地址: https://github.com/wuranxu/pity

    关注我的个人公众号测试开发坑货,催更不迷路。

  • 相关阅读:
    数据类型装换
    变量及数据类型
    27 网络通信协议 udp tcp
    26 socket简单操作
    26 socket简单操作
    14 内置函数 递归 二分法查找
    15 装饰器 开闭原则 代参装饰器 多个装饰器同一函数应用
    12 生成器和生成器函数以及各种推导式
    13 内置函数 匿名函数 eval,exec,compile
    10 函数进阶 动态传参 作用域和名称空间 函数的嵌套 全局变量
  • 原文地址:https://www.cnblogs.com/we8fans/p/15286611.html
Copyright © 2011-2022 走看看