python操作 redis 工具类（redission python）

redis 是什么及使用详情可参考官网 https://redis.io

常用数据类型（五种）

1、字符串 | 存储形式: key - value:str
    - 存储二进制数据 : 可以存储任意类型的数据，包括文本、数字、二进制数据等
    - 键值对存储 : 每个字符串有键（key）和对应的值（value）
    - 支持设置过期时间 : 可设置过期时间，到期后自动删除

 2、哈希 | 存储形式: key - { field:value, field:value ...}
    - 键值对存储 : 存储多个键值对，每个键对应一个值
    - 不可重复 : 每个key的field字段唯一，不可重复
    - 高效查找 : 通过键可以快速查找对应的值，时间复杂度为O(1)
    - 支持添加、删除、修改键值对，以及获取指定键的值

3、列表 | 存储形式: key - [value , value ...]
  - 可重复 : 同一个元素可以出现多次
  - 元素有序 : 列表中的元素按照插入顺序排列，保持有序性
  - 支持在列表两端进行元素的插入、删除操作

4、集合 | 存储形式: key - [member , member ...]
  - 不可重复 : 成员唯一，不能重复
  - 元素无序：没有固定的顺序，每次获取的元素顺序可能不同

5、有序集合 | 存储形式: key - [member:score , member:score ...]
  - 不可重复 : 成员唯一，不能重复
  - 成员按照分数的大小顺序排列，相同分数的成员按照字典序排列

特殊类型（三种）不做过多赘述，具体使用可以参考官网或者资料查询

1、geospatial 地理位置
2、hyperloglog 算法（一种概率算法，省内存）
3、bitmaps 位图

1、安装 redis

pip install redis

2、创建工具包 ，再其下创建工具类

import redis

## redis 工具类
class RedisUtil:
    def __init__(self, host='localhost', port=6379, db=0, password=''):
        pool = redis.ConnectionPool(host=host, port=port, db=db, password=password, max_connections=None)
        self.rdb = redis.Redis(connection_pool=pool)
        self.ListLeft = 0
        self.ListRight = 1

    # 程序结束后，自动关闭连接，释放资源
    def __del__(self):
        self.rdb.connection_pool.disconnect()

    # 查看 redis 版本
    def get_redis_version(self):
        version = self.rdb.info()['redis_version']
        print(f"redis server version : {version}")
        return version

    #  清除数据库所有数据
    def flush_all_data(self):
        return self.rdb.flushdb()

    # 遍历数据库所有键值对
    def print_all_data(self):
        for key in self.rdb.scan_iter(match='*'):
            key_type = self.rdb.type(key).decode('utf-8')
            if key_type == 'string':
                value = self.rdb.get(key).decode('utf-8')
            elif key_type == 'hash':
                value = self.rdb.hgetall(key)
            elif key_type == 'list':
                value = self.rdb.lrange(key, 0, -1)
            elif key_type == 'set':
                value = self.rdb.smembers(key)
            elif key_type == 'zset':
                value = self.rdb.zrange(key, 0, -1, withscores=True)
            else:
                value = "Unsupported type"

            print(f'{key_type} -> {key.decode("utf-8")} : {value}')

    # 获取指定key的类型
    def get_key_type(self, key):
        return self.rdb.type(key)

    # 判断是否存在指定key
    def is_exists_key(self, key):
        return False if self.rdb.exists(key) == 0 else True

    # 删除指定key的成员, 返回删除个数
    def delete_key(self, key):
        return self.rdb.delete(key)

    # 指定key重命名
    def rename_key(self, old, new):
        if self.is_exists_key(old):
            return self.rdb.rename(old, new)

    # 设置过期时间（以秒为单位 默认7天）
    def set_expire_by_second(self, key, second=60 * 60 * 24 * 7):
        return self.rdb.expire(key, time=second)

    # 移除key的过期时间，持久保持
    def remove_expire(self, key):
        return self.rdb.persist(key)

    # 以秒为单位返回key的剩余过期时间
    def get_expire_by_second(self, key):
        return self.rdb.ttl(key)

    # 添加字符串类型键值对
    def add_str(self, key: str, value: str):
        return self.rdb.set(key, value)

    # 获取指定key的value，若不存在key，返回 None
    def get_str(self, key):
        return self.rdb.get(key)

    # 追加指定key的value,若不存在key，新建键值对 <key:append_str(str)>
    def append_str(self, key, append_str: str):
        return self.rdb.append(key, append_str)

    # 列表添加数据  direction：添加方向（左边ListLeft /右边ListRight）
    def add_list(self, key, *value, direction=0):
        if direction == self.ListLeft:
            self.rdb.lpush(key, *value)
        elif direction == self.ListRight:
            self.rdb.rpush(key, *value)

    # 获取列表指定索引范围的列表值
    # start: 起始索引位置，end: 结束索引位置
    def get_list_by_range(self, key, start=0, end=-1):
        return self.rdb.lrange(key, start, end)

    # 获取列表指定下标的列表值
    def get_list_by_index(self, key, index):
        return self.rdb.lindex(key, index)

    # 根据删除方向删除列表数据
    # direction: 删除方向（左边ListLeft /右边ListRight）
    def pop_list_by_direction(self, key, direction=0):
        if direction == self.ListLeft:
            self.rdb.lpop(key)
        elif direction == self.ListRight:
            self.rdb.rpop(key)

    # 根据指定列表值删除列表数据
    # value:指定列表值；count:若存在多个，删除指定个数；return: 成功删除的个数
    def pop_list_by_value(self, key, value, count=1):
        result = self.rdb.lrem(key, count, value)
        return result

    # 哈希添加数据
    # field: filed域；value: value域；
    def add_hash(self, key, field, value):
        _value = self.rdb.hget(key, field)
        return self.rdb.hset(key, field, value)

    # 获取哈希指定key、field的value值
    def get_hash_by_field(self, key, field):
        return self.rdb.hget(key, field)

    # 获取所有的field值
    def get_hash_all_filed(self, key):
        return self.rdb.hkeys(key)

    # 获取所有的value值
    def get_hash_all_value(self, key):
        return self.rdb.hvals(key)

    # 获取所有的键值对
    def get_hash_all_KeyValue(self, key):
        return self.rdb.hgetall(key)

    # 集合添加数据
    def add_set(self, key, *values):
        return self.rdb.sadd(key, *values)

    # 获取集合指定key的value列表
    def get_set(self, key):
        return self.rdb.smembers(key)

    # 移除指定key的随机value值
    def delete_set_random(self, key):
        return self.rdb.spop(key)

    # 移除指定key的指定value值
    def delete_set_assign(self, key, *values):
        return self.rdb.srem(key, *values)

    # 有序集合添加数据，若已存在member，则更新分数score
    def add_sortSet(self, key, mapping):
        return self.rdb.zadd(key, mapping)

    # 获取指定下标index范围内(start,end)的成员及分数
    def get_sortSet_by_indexRange(self, key, start=0, end=-1, with_score=True):
        return self.rdb.zrange(key, start, end, withscores=with_score)

    # 获取指定分数score范围内(min,max)的成员及分数
    def get_sortSet_by_scoreRange(self, key, min, max):
        return self.rdb.zrangebyscore(key, min, max)

    # 删除指定member的成员
    def delete_sortSet_by_member(self, key, *members):
        return self.rdb.zrem(key, *members)

    # 删除指定下标index范围内(start,end)的成员及分数
    def delete_sortSet_by_indexRange(self, key, start=0, end=-1):
        return self.rdb.zremrangebyrank(key, start, end)

    # 删除指定分数score范围内(min,max)的成员及分数
    def delete_sortSet_by_scoreRange(self, key, min, max):
        return self.rdb.zremrangebyscore(key, min, max)

3、使用工具类测试

from utils.RedisUtil import RedisUtil

client = RedisUtil()
# 查询版本
# client.get_redis_version()
# TODO:获取所有数据，线上不可使用
# client.print_all_data()
# 获取指定 key 类型
# result = client.get_key_type("k1")


# 判断是否存在
# result = client.is_exists_key("k1")
# 删除 key
# result = client.delete_key("k1")

# 重命名
# result = client.rename_key("k2", "k1")

result = client.get_expire_by_second("k1")
print(result)