Python - Python操作Redis

安装Redis可参考

Redis-入门简介-CSDN博客

在Python中接入Redis数据库通常使用redis-py这个库

一、安装Redis 

首先，需要安装redis-py库。通过pip来安装

pip install redis

二、连接Redis

'''
Redis连接操作
'''
import redis


def redis_connect():
    try:
        redisClient = redis.Redis(host='127.0.0.1', port=6379, password=None)
        # <redis.client.Redis(<redis.connection.ConnectionPool(<redis.connection.Connection(host=127.0.0.1,port=6379,db=0)>)>)>
        print(redisClient)

        # 设置key为‘name’的值以及过期时间
        redisClient.set('name', 'zhangsan', ex=300)
        # 获取key为‘name’的过期时间
        print(redisClient.ttl('name'))  # 300
        # 获取key为‘name’的值
        print(redisClient.get('name'))  # b'zhangsan';前面的'b'代表字节

        redisClient.set('nickname', '张三', ex=500)
        print(redisClient.get('nickname'))  # b'\xe5\xbc\xa0\xe4\xb8\x89'
        print(redisClient.get('nickname').decode())  # 张三

        redisClient.set('num', '10')
        print(redisClient.incr('num'))  # 11
        print(redisClient.incrby('num', 5))  # 16
        print(int(redisClient.get('num').decode()))  # 16

        # 使用哈希表
        redisClient.hset('user', 'id', '1001')
        redisClient.hset('user', 'name', '张三')
        print(redisClient.hgetall('user'))  # {b'id': b'1001', b'name': b'\xe5\xbc\xa0\xe4\xb8\x89'}
        print(redisClient.hget('user', 'name').decode())  # 张三

        # 使用列表
        redisClient.lpush('list', 'a', 'b', 'c')
        # 获取列表中的所有元素
        print(redisClient.lrange('list', 0, -1))  # [b'c', b'b', b'a']

        # 使用集合
        redisClient.sadd('myset', 'a')
        redisClient.sadd('myset', 'b')
        redisClient.smembers('myset')  # 获取集合中的所有元素

    except Exception as ex:
        print(ex.args)
    finally:
        redisClient.close()

if __name__ == '__main__':
    redis_connect()

使用连接池（可选）

为了提高性能和效率，特别是在多线程环境下，建议使用连接池

'''
Redis连接池
'''
import redis

def redis_connect():
    """
    redis连接
    :return:
    """
    try:
        pool = redis.ConnectionPool(host='127.0.0.1', port=6379, decode_responses=True)
        redisClient = redis.Redis(connection_pool=pool)
        redisClient.set('nickname', '张三', ex=500)
        print(redisClient.get('nickname'))  # b'\xe5\xbc\xa0\xe4\xb8\x89'
        print(redisClient.get('nickname').decode())  # 张三
    except Exception as ex:
        print(ex)
    finally:
        redisClient.close()
        pool.close()

if __name__ == '__main__':
    redis_connect()

在使用连接池的情况下，通常不需要手动关闭每个连接，连接池会自动管理连接。如果你使用的是连接池，只需在不再需要时关闭整个连接池

    finally:
        redisClient.close()
        pool.close()

三、封装连接Redis数据库工具

'''
1、初始化连接：在 __init__ 中，我们通过 redis.StrictRedis 初始化 Redis 连接，可以根据需要传递不同的 Redis 服务器信息。
2、基本操作封装：
（1）set 和 get 分别用于设置和获取键值。
（2）delete 用于删除指定的键。
（3）exists 用于检查键是否存在。
3、批量操作：提供了 set_multiple 和 get_multiple 方法来处理批量的 Redis 操作。
4、自增和自减：封装了 increment 和 decrement 方法来处理数字类型的键值的增减。
5、哈希表操作：通过 hset, hget, 和 hgetall 封装了对 Redis 哈希表的操作
'''

import redis
from typing import Optional, Union, Any


class RedisHelper:
    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        """
        初始化 RedisHelper 类，设置 Redis 连接信息
        :param host: Redis 服务器主机，默认是 localhost
        :param port: Redis 服务器端口，默认是 6379
        :param db: Redis 数据库，默认是 0
        """
        self.redis_client = redis.StrictRedis(host=host, port=port, db=db, decode_responses=True)

    def set(self, key: str, value: Union[str, int, float], ex: Optional[int] = None) -> bool:
        """
        设置键值对，如果设置了 ex 参数，则键值会在过期时间后自动删除
        :param key: 键
        :param value: 值，可以是字符串、整数或浮点数
        :param ex: 过期时间（秒），默认不设置过期时间
        :return: 是否成功
        """
        try:
            if ex:
                self.redis_client.setex(key, ex, value)
            else:
                self.redis_client.set(key, value)
            return True
        except Exception as e:
            print(f"Error setting key {key}: {e}")
            return False

    def get(self, key: str) -> Optional[str]:
        """
        获取指定键的值
        :param key: 键
        :return: 键对应的值，如果键不存在，则返回 None
        """
        try:
            return self.redis_client.get(key)
        except Exception as e:
            print(f"Error getting key {key}: {e}")
            return None

    def delete(self, key: str) -> bool:
        """
        删除指定键
        :param key: 键
        :return: 是否成功
        """
        try:
            self.redis_client.delete(key)
            return True
        except Exception as e:
            print(f"Error deleting key {key}: {e}")
            return False

    def exists(self, key: str) -> bool:
        """
        检查指定的键是否存在
        :param key: 键
        :return: 键是否存在
        """
        try:
            return self.redis_client.exists(key)
        except Exception as e:
            print(f"Error checking existence of key {key}: {e}")
            return False

    def set_multiple(self, mapping: dict, ex: Optional[int] = None) -> bool:
        """
        批量设置多个键值对
        :param mapping: 键值对字典
        :param ex: 过期时间（秒），可选
        :return: 是否成功
        """
        try:
            if ex:
                for key, value in mapping.items():
                    self.redis_client.setex(key, ex, value)
            else:
                self.redis_client.mset(mapping)
            return True
        except Exception as e:
            print(f"Error setting multiple keys: {e}")
            return False

    def get_multiple(self, keys: list) -> dict:
        """
        批量获取多个键的值
        :param keys: 键列表
        :return: 键值对字典
        """
        try:
            values = self.redis_client.mget(keys)
            return dict(zip(keys, values))
        except Exception as e:
            print(f"Error getting multiple keys: {e}")
            return {}

    def increment(self, key: str, amount: int = 1) -> Union[int, None]:
        """
        对指定键的值进行自增
        :param key: 键
        :param amount: 增量，默认为 1
        :return: 增加后的值，如果操作失败，则返回 None
        """
        try:
            return self.redis_client.incrby(key, amount)
        except Exception as e:
            print(f"Error incrementing key {key}: {e}")
            return None

    def decrement(self, key: str, amount: int = 1) -> Union[int, None]:
        """
        对指定键的值进行自减
        :param key: 键
        :param amount: 减量，默认为 1
        :return: 减少后的值，如果操作失败，则返回 None
        """
        try:
            return self.redis_client.decrby(key, amount)
        except Exception as e:
            print(f"Error decrementing key {key}: {e}")
            return None

    def hset(self, hash_name: str, key: str, value: Any) -> bool:
        """
        向哈希表中设置字段值
        :param hash_name: 哈希表名称
        :param key: 字段名
        :param value: 字段值
        :return: 是否成功
        """
        try:
            self.redis_client.hset(hash_name, key, value)
            return True
        except Exception as e:
            print(f"Error setting hash {hash_name}:{key}: {e}")
            return False

    def hget(self, hash_name: str, key: str) -> Optional[Any]:
        """
        获取哈希表中的字段值
        :param hash_name: 哈希表名称
        :param key: 字段名
        :return: 字段值，如果字段不存在，则返回 None
        """
        try:
            return self.redis_client.hget(hash_name, key)
        except Exception as e:
            print(f"Error getting hash {hash_name}:{key}: {e}")
            return None

    def hgetall(self, hash_name: str) -> dict:
        """
        获取哈希表中的所有字段和值
        :param hash_name: 哈希表名称
        :return: 键值对字典
        """
        try:
            return self.redis_client.hgetall(hash_name)
        except Exception as e:
            print(f"Error getting all hash fields for {hash_name}: {e}")
            return {}


# 示例使用
if __name__ == '__main__':
    redis_helper = RedisHelper()

    # 设置单个键值对
    redis_helper.set('name', 'Allen')

    # 获取单个键值
    print(redis_helper.get('name')) # Allen

    # 设置多个键值对
    redis_helper.set_multiple({'age': 18, 'location': 'shanghai'})

    # 获取多个键值
    print(redis_helper.get_multiple(['name', 'age', 'location'])) # {'name': 'Allen', 'age': '18', 'location': 'shanghai'}

    # 增加某个键的值
    redis_helper.increment('age')
    print(redis_helper.get('age')) # 19

    # 设置哈希表
    redis_helper.hset('user:1000', 'name', 'Allen')
    print(redis_helper.hget('user:1000', 'name')) # Allen

1、初始化连接：在 __init__ 中，我们通过 redis.StrictRedis 初始化 Redis 连接，可以根据需要传递不同的 Redis 服务器信息。

2、基本操作封装：

（1）set 和 get 分别用于设置和获取键值。

（2）delete 用于删除指定的键。

（3）exists 用于检查键是否存在。

3、批量操作：提供了 set_multiple 和 get_multiple 方法来处理批量的 Redis 操作。

4、自增和自减：封装了 increment 和 decrement 方法来处理数字类型的键值的增减。

5、哈希表操作：通过 hset, hget, 和 hgetall 封装了对 Redis 哈希表的操作