测试平台系列(79) 编写Redis配置功能(下)

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 编写Redis配置功能(下)

大家好~我是米洛


我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的完整教程,希望大家多多支持。


欢迎关注我的龚仲耗测试开发坑货,获取最新文章教程!

回顾


上一节我们提出了优化Dao逻辑的想法,那今天就试着来兑现之,并运用到Redis配置管理的开发中去。

初步构思list方法


我们在dao/init.py新建类: Mapper,以后所有的dao类都继承自它。

想想list需要什么,一般需要,字段参数是like还是等于这3个重要的信息。

明白这个以后,我们的伪代码就好编写了:


# 1. 获取session
async with async_session() as session:
  condition = [model.deleted_at == 0]
  # 2. 根据参数里的字段,字段值构造查询条件
  DatabaseHelper.where(字段,字段值,condition)
  # 3. 查询出结果,后续与原来的方式一致,就不写了

可以看到,这边除了需要上面的3个信息以外,还需要try去写入日志,也就是说需要我们平时经常创建的log成员变量,还需要model这个类,否则你不知道改的是什么表。

思考


我们平时的参数,都是key-value的形式,一旦value不为空,那么说明我们要根据这个条件来查询。

而model和log,我们可以通过类的装饰器,由子类传递给父类(这里会用到setattr和getattr)


class Mapper(object):
    log = None
    model = None
    @classmethod
    async def list_data(cls, **kwargs):
        try:
            async with async_session() as session:
                # 构造查询条件,默认是数据未被删除
                condition = [getattr(cls.model, "deleted_at") == 0]
                # 遍历参数,当参数不为None的时候传递
                for k, v in kwargs.items():
                    # 判断是否是like的情况 TODO: 这里没支持in查询
                    like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
                    # 如果是like模式,则使用Model.字段.like 否则用 Model.字段 等于
                    DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
                                         condition)
                result = await session.execute(select(cls.model).where(*condition))
                return result.scalars().all()
        except Exception as e:
            # 这边调用cls本身的log参数,写入日志+抛出异常
            cls.log.error(f"获取{cls.model}列表失败, error: {e}")
            raise Exception(f"获取数据失败")

注释写的比较详细,由于现在字段是个变量,所以我们不能用model.字段来取值,所以取而代之的是getattr(model, 字段)。不熟悉的朋友可以去搜索下getattr

这样,一个粗略的list方法就写好了,但是这个是不带分页的,所以我们还需要补充一个分页的模式,其实也很简单。


@classmethod
    async def list_data_with_pagination(cls, page, size, **kwargs):
        try:
            async with async_session() as session:
                condition = [getattr(cls.model, "deleted_at") == 0]
                for k, v in kwargs.items():
                    like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
                    sql = DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
                                               condition)
                return await DatabaseHelper.pagination(page, size, session, sql)
        except Exception as e:
            cls.log.error(f"获取{cls.model}列表失败, error: {e}")
            raise Exception(f"获取数据失败")

基本上长的差不多,只是最后返回那里,调用了pagination相关方法。

看看dao装饰器


5.jpg

这也是为什么要用classmethod而不是staticmethod的原因

我们很坏,把这个装饰器套到子类上,并把model和log传给父类。毕竟方法都是在调用父类的方法,父类无法直接拿到子类的数据

6.jpg

用的话,就是把model和log传入dao参数

这样就避免了在调用list方法的时候,还需要传入model和log的尴尬情况。

完善其他方法


list搞定以后,其他的还会远吗?但还真的好像还有点问题,因为我们一般会有一些重名判断,但没关系,我们可以编写一个query方法。


@classmethod
    def query_wrapper(cls, **kwargs):
        condition = [getattr(cls.model, "deleted_at") == 0]
        # 遍历参数,当参数不为None的时候传递
        for k, v in kwargs.items():
            # 判断是否是like的情况 TODO: 这里没支持in查询
            like = v is not None and len(v) > 2 and v.startswith("%") and v.endswith("%")
            # 如果是like模式,则使用Model.字段.like 否则用 Model.字段 等于
            DatabaseHelper.where(v, getattr(cls.model, k).like(v) if like else getattr(cls.model, k) == v,
                                 condition)
        return select(cls.model).where(*condition)
    @classmethod
    async def query_record(cls, **kwargs):
        try:
            async with async_session() as session:
                sql = cls.query_wrapper(**kwargs)
                result = await session.execute(sql)
                return result.scalars().first()
        except Exception as e:
            cls.log.error(f"查询{cls.model}失败, error: {e}")
            raise Exception(f"查询数据失败")

由于查询方法太过于通用,所以抽成了query_wrapper方法。

  • 正常编写insert接口


@classmethod
    async def insert_record(cls, model):
        try:
            async with async_session() as session:
                async with session.begin():
                    session.add(model)
                    await session.flush()
                    session.expunge(model)
                    return model
        except Exception as e:
            cls.log.error(f"添加{cls.model}记录失败, error: {e}")
            raise Exception(f"添加记录失败")

这边返回了model,如果不需要也可以不用,但咱还是给返回。

  • insert接口层

7.jpg

image

和以前一样,先查,如果没有再关掉。

但这样会产生2个session,开->关->开->关

但也解耦了查询和插入2个操作。

  • 编写删除和修改方法


@classmethod
    async def update_record_by_id(cls, user, model, not_null=False):
        try:
            async with async_session() as session:
                async with session.begin():
                    query = cls.query_wrapper(id=model.id)
                    result = await session.execute(query)
                    original = result.scalars().first()
                    if original is None:
                        raise Exception("记录不存在")
                    DatabaseHelper.update_model(original, model, user, not_null)
                    await session.flush()
                    session.expunge(original)
                    return original
        except Exception as e:
            cls.log.error(f"更新{cls.model}记录失败, error: {e}")
            raise Exception(f"更新记录失败")
    @classmethod
    async def delete_record_by_id(cls, user, id):
        """
        逻辑删除
        :param user:
        :param id:
        :return:
        """
        try:
            async with async_session() as session:
                async with session.begin():
                    query = cls.query_wrapper(id=id)
                    result = await session.execute(query)
                    original = result.scalars().first()
                    if original is None:
                        raise Exception("记录不存在")
                    DatabaseHelper.delete_model(original, user)
        except Exception as e:
            cls.log.error(f"删除{cls.model}记录失败, error: {e}")
            raise Exception(f"删除记录失败")
    @classmethod
    async def delete_by_id(cls, id):
        """
        物理删除
        :param id:
        :return:
        """
        try:
            async with async_session() as session:
                async with session.begin():
                    query = cls.query_wrapper(id=id)
                    result = await session.execute(query)
                    original = result.scalars().first()
                    if original is None:
                        raise Exception("记录不存在")
                    session.delete(original)
        except Exception as e:
            cls.log.error(f"逻辑删除{cls.model}记录失败, error: {e}")
            raise Exception(f"删除记录失败")

删除这边支持了物理删除和逻辑删除,当然我们一般是用软删除

完善其他接口


8.jpg

image

可以看到删除和修改和以前是差不多的(也是内部进行数据是否存在判断),这些步骤做完。redis的管理工作就可以顺利进行了,接着我们需要为之编写页面咯!

由于是很基础的表格页面,所以我们不赘述。下一节我们会利用配置的redis连接编写redisManager,管理我们的连接数据,为之后在线执行redis以及将它作为前置条件打下基础。



相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
打赏
0
0
0
0
6
分享
相关文章
Docker平台上的Redis镜像运行
这就是如何在Docker平台上运行Redis镜像的全部过程。走进Docker和Redis的世界,探索更多可能!
85 10
os-copilot安装_配置_功能测试全集
我是一位中级运维工程师,我平时工作会涉及到 各类服务器的 数据库 与 java环境配置 操作。 我顺利使用了OS Copilot的 -t -f | 功能,我的疑惑是不能在自动操作过程中直接给与脚本运行权限,必须需要自己运行一下 chmod 这个既然有了最高的权限,为什么就不能直接给与运行权限呢。 我认为 -t 功能有用,能解决后台运行基础命令操作。 我认为 -f 功能有用,可以通过task文件中撰写连续任务操作。 我认为 | 对文件理解上有很直接的解读,可以在理解新程序上有很大帮助。
193 86
os-copilot使用之全面配置与使用测试
作为一名个人开发者,我主要从事云服务器架设工作。近期,我成功使用了OS Copilot的 `-t -f |` 功能,解决了执行语句、连续提问及快速理解文件的问题。我发现这些功能非常实用,特别是在使用Workbench时能快速调用AI助手。此外,建议将AI功能与xShell工具联动,进一步提升效率。文中详细记录了购买服务器、远程连接、安装配置OS Copilot以及具体命令测试的过程,展示了如何通过快捷键和命令行操作实现高效开发。
186 67
【01】噩梦终结flutter配安卓android鸿蒙harmonyOS 以及next调试环境配鸿蒙和ios真机调试环境-flutter项目安卓环境配置-gradle-agp-ndkVersion模拟器运行真机测试环境-本地环境搭建-如何快速搭建android本地运行环境-优雅草卓伊凡-很多人在这步就被难倒了
【01】噩梦终结flutter配安卓android鸿蒙harmonyOS 以及next调试环境配鸿蒙和ios真机调试环境-flutter项目安卓环境配置-gradle-agp-ndkVersion模拟器运行真机测试环境-本地环境搭建-如何快速搭建android本地运行环境-优雅草卓伊凡-很多人在这步就被难倒了
322 3
【01】噩梦终结flutter配安卓android鸿蒙harmonyOS 以及next调试环境配鸿蒙和ios真机调试环境-flutter项目安卓环境配置-gradle-agp-ndkVersion模拟器运行真机测试环境-本地环境搭建-如何快速搭建android本地运行环境-优雅草卓伊凡-很多人在这步就被难倒了
在 Ubuntu 20.04 上安装和配置 Redis
在 Ubuntu 20.04 上安装和配置 Redis 的步骤如下:首先更新系统包,然后通过 `apt` 安装 Redis。安装后,启用并启动 Redis 服务,检查其运行状态。可选配置包括修改绑定 IP、端口等,并确保防火墙设置允许外部访问。最后,使用 `redis-cli` 测试 Redis 功能,如设置和获取键值对。
67 1
Redis 功能扩展 Lua 脚本 对Redis扩展 eval redis.call redis.pcall
通过本文的介绍,我们详细讲解了 Lua 脚本在 Redis 中的作用、`eval` 命令的使用方法以及 `redis.call` 和 `redis.pcall` 的区别和用法。通过合理使用 Lua 脚本,可以实现复杂的业务逻辑,确保操作的原子性,并减少网络开销,从而提高系统的性能和可靠性。
121 13
【02】写一个注册页面以及配置打包选项打包安卓apk测试—开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草央千澈
【02】写一个注册页面以及配置打包选项打包安卓apk测试—开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草央千澈
81 1
【02】写一个注册页面以及配置打包选项打包安卓apk测试—开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草央千澈
OS-Copilot参数功能全面测试报告
作为一名运维工程师,我主要负责云资源的运维和管理。通过使用OS Copilot的-t/-f/管道功能,我顺利解决了环境快速搭建的问题,例如Tomcat的快速部署。具体步骤包括购买ECS服务器、配置安全组、远程登录并安装OS Copilot。使用-f参数成功安装并启动Tomcat,自动配置JDK,并通过|管道功能验证了生成内容的正确性。整个过程非常流畅,极大提升了工作效率。
80 12
NoSQL与Redis配置与优化
通过合理配置和优化Redis,可以显著提高其性能和可靠性。选择合适的数据结构、优化内存使用、合理设置持久化策略、使用Pipeline批量执行命令、以及采用分布式集群方案,都是提升Redis性能的重要手段。同时,定期监控和维护Redis实例,及时调整配置,能够确保系统的稳定运行。希望本文对您在Redis的配置与优化方面有所帮助。
99 23

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等