redis实操-查询数据库信息存入redis对比两次差异

怪盗LYL

发布于 2022-06-13 11:22:13

6880

发布于 2022-06-13 11:22:13

文章被收录于专栏：测试开发真货

我们来看下总体效果，数据库是这样的，第一次运行判断redis里面没有关键字作为key。将数据以关键字作为key入到redis里。

再次运行查询到redis里有关键字作为key的数据，判断两次差异。

Redis 安装：

下载地址：https://github.com/tporadowski/redis/releases。下载 Redis-x64-xxx.zip并解压，双击运行redis-server.exe即可在本地运行一个默认redis服务，默认无密码。效果如下：

import json
import redis
from sqlalchemy import create_engine
import pandas as pd

用到redis模块操作redis连接sqlalchemy 模块连接mysql数据库，pandas 模块可解析数据，json模块转换数据为json格式。

def getredis(keystr,valuestr):
       # r = redis.Redis(host='127.0.0.1', password='密码', port=端口号)使用密码的场景
       r = redis.Redis(host='127.0.0.1', port=6379) #连接redis
       keystr = keystr #获取想用的key值
       if r.get(keystr): #如果通过key存在
           value = r.get(keystr).decode('utf8') #获取value值并解码
           print("更新前：" + value)
           print("更新后：" + valuestr)
           if value == valuestr: #判断是否有改动
               print("无改动")
           else:
               r.set(keystr, valuestr) #讲值塞入redis value值为数据库查询出的数据
               afterstr = json.loads(valuestr) #转为json格式用来判断
               beforestr = json.loads(value)
               for x in beforestr:
                   if beforestr[x] == afterstr[x]: #判断是否相等
                       pass
                   else:
                       print(x, "从", beforestr[x], '变为了', afterstr[x])
       else:
           print("该订单无redis数据")
           r.set(keystr, valuestr)
           print("塞值:" + valuestr)

写好判断和存取redis的函数后就可以写查询数据库的函数了。

def select(db,sql):
    engine = db
    sql = sql
    # read_sql_query的两个参数: sql语句， 数据库连接
    df = pd.read_sql_query(sql, engine) #使用pandas查询数据库
    return df
def basedb():
    Test = create_engine("mysql+pymysql://账号:密码@ip:端口/数据库", #数据库连接信息
                             echo=True)
    return Test
def getdate(keystr,table): #入口
    db=basedb() #获取y一个数据库连接
    sql=f"select * from {table} where stu_id='{keystr}'" #准备好sql语句
    df=select(db,sql) #执行sql的函数
    try:
        jilu = df.iloc[0] #选择第一条记录信息，基于索引位来选取数据集
        valuestr = jilu.to_json(orient='index', force_ascii=False)  # 按照index转换成json格式 字符串
        # valuestr=json.dumps(valuestr,ensure_ascii=False)
        return valuestr
    except Exception as e:
        print("无值或者获取失败")
        return False

主要调用：

if __name__ == '__main__':
    keystr='1001'
    table='students'
    valuestr=getdate(keystr,table)
    if  valuestr:
        getredis(keystr=keystr,valuestr=valuestr)

我这里是手动运行了两次，读者可以根据自己需要加到自己的测试用例初始化和测试完成后的断言两个地方来判断。

没想好明天水啥。各位有需要可以留言。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-08-10，如有侵权请联系 cloudcommunity@tencent.com 删除

json