code-segment/db_helper/redis_helper.py

# redis数据库相关操作
import redis


class RedisHelper(object):
    def __init__(self, redis_info):
        """
        初始化redis连接信息
        :param redis_info: redis连接信息, 格式：dict, {'host': '', 'port': '', 'password': ''}
        """
        self.host = redis_info['host']
        self.port = redis_info['port']
        self.password = redis_info['password']

    def connect(self):
        """
        连接redis
        :return: conn
        """
        # 创建连接池
        pool = redis.ConnectionPool(
            host=self.host,
            port=self.port,
            password=self.password,
            decode_responses=True
        )
        # 创建redis对象
        conn = redis.Redis(connection_pool=pool)
        return conn

    def key_exists(self, key_name):
        """
        判断key是否存在
        :param key_name: key
        :return: 存在-1, 不存在-0
        """
        conn = self.connect()
        return conn.exists(key_name)

    def del_keys(self, key_name_list):
        """
        删除key
        :param key_name_list: keys, type-list
        :return: None
        """
        conn = self.connect()
        conn.delete(*key_name_list)

    def set_data_to_redis(self, key_name, value, expire_time=24*3600):
        """
        新增数据
        :param key_name: key
        :param value: 元素的值
        :param expire_time: 过期时间，单位：s，默认1天
        :return: None
        """
        conn = self.connect()
        conn.set(key_name, value, ex=expire_time)


if __name__ == '__main__':
    REDIS_INFO = {
        'host': 'r-bp1ps6my7lzg8rdhwx682.redis.rds.aliyuncs.com',
        'port': 6379,
        'password': 'Wqsd@2019',
    }
    redis_helper = RedisHelper(redis_info=REDIS_INFO)
    ex = redis_helper.key_exists('com.weiqu.video.recall.hot.item.score.20211115')
    print(ex)
    redis_helper.del_keys(['com.weiqu.video.recall.hot.item.score.20211112', '1'])