Django 集成Redis缓存库

不念不忘少年蓝@ 2023-10-10 13:13 44阅读 0赞

## 第一步:Python 安装redis 模块 ##

pip install redis

## 第二步：封装Redis 功能模块(string\\hash\\list) ##

import redis
    import logging
    
    # 日志记录初始化
    logger = logging.getLogger('mylogger')
    
    
    # redis 工具类封装
    class RedisUtil:
        def __init__(self, ip, passwd, port=6379, db=0):
            try:
                self.r = redis.Redis(host=ip, password=passwd, port=port, db=db, decode_responses=True, charset='UTF-8')
            except Exception as e:
                print('redis连接失败，错误信息%s' % e)
                logger.error('redis连接失败，错误信息%s' % e)
    
        def str_get(self, k):
            res = self.r.get(k)
            if res:
                return res.decode()
    
        def str_set(self, k, v, time=None):
            self.r.set(k, v, time)
    
        def delete(self, k):
            tag = self.r.exists(k)  # 判断这个Key是否存在
            if tag:
                self.r.delete(k)
                print('删除成功')
            else:
                print('这个key不存在')
    
        def hash_get(self, name, key):
            res = self.r.hget(name, key)
            if res:
                return res.decode()
    
        def hash_set(self, name, k, v):
            self.r.hset(name, k, v)
    
        def hash_all(self, name):
            res = self.r.hgetall(name)
            new_dict = {}
            if res:
                for k, v in res.items():
                    k = k.decode()
                    v = v.decode()
                    new_dict[k] = v
            return new_dict
    
        def hash_del(self, name, k):
            res = self.r.hdel(name, k)
            if res:
                print('删除成功')
                return True
            else:
                print('删除失败.该key不存在')
                return False
    
        def list_set(self, name, value):
            self.r.lpush(name, value)
    
        def list_range(self, name, start, end):
            new_list = []
            res = self.r.lrange(name, start, end)
            if res:
                for value in res:
                    new_list.append(value)
            return new_list
    
        def list_len(self, name):
            return self.r.llen(name)
    
        def list_remove(self, name, count, value):
            self.r.lrem(name, count, value)
    
        def list_index(self, name, index):
            self.r.lindex(name, index)
    
        @property
        def clean_redis(self):
            self.r.flushdb()  # 清空redis
            print('清空redis成功.')
            return 0

备注说明:在应用userSystem中，新建redis\_tool.py 文件，新增如上代码。

## 第三步：引用redis 工具类，测试封装功能代码 ##

在应用userSystem中，在views.py 文件中添加redis\_tool.py 文件依赖

from userSystem.redis_tool import RedisUtil

实例化RedisUtil类，测试相关方法，相关功能验证代码如下:

# redis 功能验证
    def redisStr(request):
        redis_util = RedisUtil(ip='192.168.1.73', passwd='abc123')
        json_str = request.body
        json_str = json_str.decode()  # python3.6及以上不用这一句代码
        dict_data = json.loads(json_str)  # loads把str转换为dict，dumps把dict转换为str
        # redis字符串操作
        # redis_str = dict_data.get('redisstr')
        # 字符串新增
        # redis_util.str_set(redis_str, redis_str)
        # 字符串查询
        # print(redis_util.str_get(redis_str))
        # 字符串删除
        # redis_util.delete(redis_str)
    
        # redis Hash操作
        # redis_hash = 'hash'
        # objDict = dict_data.get(redis_hash)
        # 数据类型为字典dict
        # print(type(objDict))
        # 遍历字典，hash 数据填充(redis Hash新增)
        # for k, v in objDict.items():
            # redis_util.hash_set(redis_hash, k, v)
        # redis hash 查询(name-key)
        # value = redis_util.hash_get(redis_hash, 'key')
        # print(value)
        # redis hash 查询(name)
        # dicts = redis_util.hash_all(redis_hash)
        # print(dicts)
        # redis hash 删除(name-key)
        # redis_util.hash_del(redis_hash, 'key')
        # redis hash 通用删除(name)
        # redis_util.delete(redis_hash)
    
        # redis List 操作
        redis_list = 'list'
        obj_str = dict_data.get(redis_list)
        # redis List 新增
        redis_util.list_set(redis_list, redis_list)
        # 传递参数为数组,并追加至redis List 集合中
        obj_array = dict_data.get("obj_array")
        # 数据类型为list
        print(type(obj_array))
        for value in obj_array:
            # redis list 集合追加
            redis_util.list_set(redis_list, value)
    
        # redis list 访问查询
        list = redis_util.list_range(redis_list, 0, 4)
        print(list)
        # redis list 长度
        print(redis_util.list_len(redis_list))
        # redis list 移除
        print(redis_util.list_remove(redis_list, 0, 'list'))
        # redis list 查询指定下标元素
        print(redis_util.list_index(redis_list, 1))
    
    
    
        return response_success(message='redis 字符串操作')

项目截图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210812120926854.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob3V6aGl3ZW5nYW5n,size_16,color_FFFFFF,t_70