robots-auto-action/tools/redis_util.py

"""
redis工具类
"""
import redis

from tools import redis_pool


class redis_util(object):

    def __init__(self):
        # 连接redis
        # self.__redis = redis.StrictRedis(host=host, password=password, port=port)
        self.__redis = redis_pool.redis_pool.get_conn()

    # 设置key-value
    def set(self, key, value, timeout=6 * 60 * 60):
        self.__redis.set(key, value)
        # key超时时间
        self.__redis.expire(name=key, time=timeout)

    # 获取key-value
    def get(self, key):
        return self.__redis.get(key)

    # 删除key-value
    def delete(self, key):
        return self.__redis.delete(key)

    # 判断key是否存在
    def is_exists_key(self, key):
        # 返回1存在，0不存在
        return self.__redis.exists(key)

    # 添加集合操作
    def add_set(self, key, value):
        # 集合中存在该元素则返回0,不存在则添加进集合中，并返回1
        # 如果key不存在，则创建key集合，并添加元素进去,返回1
        return self.__redis.sadd(key, value)

    # 判断value是否在key集合中
    def is_insert(self, key, value):
        """判断value是否在key集合中，返回布尔值"""
        return self.__redis.sismember(key, value)

    # 模糊匹配key_prefix
    def match_pattern_prefix(self, key_prefix):
        """模糊匹配key_prefix"""
        return self.__redis.keys(pattern=f"{key_prefix}*")

    # 模糊匹配key_suffix
    def match_pattern_suffix(self, key_suffix):
        """模糊匹配key_suffix"""
        return self.__redis.keys(pattern=f"*{key_suffix}")

    # 模糊匹配key
    def match_pattern_key(self, key):
        """模糊匹配key"""
        return self.__redis.keys(pattern=f"*{key}*")

    # 自增
    def incr(self, name, amount):
        """模糊匹配key"""
        return self.__redis.incr(name, amount)

    # 对list类型进行push
    def left_push(self, name, value):
        """模糊匹配key"""
        return self.__redis.lpush(name, value)

    # 获取list 对应key
    def lrange(self, key, start_index, end_index):
        return self.__redis.lrange(key, start_index, end_index)

    # 获取list 对应key
    def set_nx(self, key, value, timeout=6 * 60 * 60):
        set_true = self.__redis.setnx(key, value)
        if set_true:
            self.__redis.expire(name=key, time=timeout)
        return set_true

    # 关闭连接
    def close(self):
        self.__redis.close()