1、redis-py 库封装一个 Redis 工具类可以帮助我们简化 Redis 的操作并提高代码的复用性和可维护性。

• 安装redis

pip install redis

import redis
import logginglogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class RedisUtils:def __init__(self, host='localhost', port=6379, db=0, decode_responses=True,password=None):"""初始化 Redis 连接:param host: Redis 主机，默认为 localhost:param port: Redis 端口，默认为 6379:param db: Redis 数据库，默认为 0:param decode_responses: 如果为 True，则自动解码 Redis 返回的字节串为字符串:param password: Redis密码（如果设置了密码）"""self.host = hostself.port = portself.db = dbself.decode_responses = decode_responses# 创建 Redis 连接实例self.redis = redis.StrictRedis(host=self.host,port=self.port, db=self.db, decode_responses=self.decode_responses,password=password)def set(self, key, value, ex=None, px=None, nx=False, xx=False):"""设置 Redis 键值对:param key: 键:param value: 值:param ex: 键值对过期时间（秒）:param px: 键值对过期时间（毫秒）:param nx: 只在键不存在时设置:param xx: 只在键存在时设置:return: 执行结果（True 或 False）"""try:result = self.redis.set(key, value, ex=ex, px=px, nx=nx, xx=xx)logger.info(f"Set key '{key}' with value '{value}'")return resultexcept Exception as e:logger.error(f"Error setting key '{key}': {e}")return Nonedef get(self, key):"""获取 Redis 键的值:param key: 键:return: 值（如果不存在则返回 None）"""try:value = self.redis.get(key)logger.info(f"Get key '{key}' with value '{value}'")return valueexcept Exception as e:logger.error(f"Error getting key '{key}': {e}")return Nonedef delete(self, key):"""删除 Redis 键:param key: 键:return: 删除结果（删除的键的数量）"""try:result = self.redis.delete(key)logger.info(f"Deleted key '{key}'")return resultexcept Exception as e:logger.error(f"Error deleting key '{key}': {e}")return Nonedef exists(self, key):"""判断 Redis 键是否存在:param key: 键:return: 是否存在（True 或 False）"""try:exists = self.redis.exists(key)logger.info(f"Key '{key}' exists: {exists}")return existsexcept Exception as e:logger.error(f"Error checking existence of key '{key}': {e}")return Falsedef hset(self, name, key, value):"""设置哈希表中的字段:param name: 哈希表名:param key: 字段名:param value: 字段值:return: 执行结果"""try:result = self.redis.hset(name, key, value)logger.info(f"Set hash '{name}' field '{key}' with value '{value}'")return resultexcept Exception as e:logger.error(f"Error setting hash '{name}' field '{key}': {e}")return Nonedef hget(self, name, key):"""获取哈希表中的字段:param name: 哈希表名:param key: 字段名:return: 字段值"""try:value = self.redis.hget(name, key)logger.info(f"Get hash '{name}' field '{key}' with value '{value}'")return valueexcept Exception as e:logger.error(f"Error getting hash '{name}' field '{key}': {e}")return Nonedef lpush(self, name, *values):"""向 Redis 列表左侧插入元素:param name: 列表名:param values: 要插入的值:return: 执行结果（列表长度）"""try:result = self.redis.lpush(name, *values)logger.info(f"LPush to list '{name}' values {values}")return resultexcept Exception as e:logger.error(f"Error pushing to list '{name}': {e}")return Nonedef lrange(self, name, start, end):"""获取 Redis 列表中的元素:param name: 列表名:param start: 起始索引:param end: 结束索引:return: 列表元素"""try:values = self.redis.lrange(name, start, end)logger.info(f"LRANGE from list '{name}' from index {start} to {end} with values {values}")return valuesexcept Exception as e:logger.error(f"Error getting range from list '{name}': {e}")return Nonedef sadd(self, name, *values):"""向 Redis 集合添加元素:param name: 集合名:param values: 要添加的值:return: 添加的元素数量"""try:result = self.redis.sadd(name, *values)logger.info(f"SAdd to set '{name}' values {values}")return resultexcept Exception as e:logger.error(f"Error adding to set '{name}': {e}")return Nonedef smembers(self, name):"""获取 Redis 集合中的所有元素:param name: 集合名:return: 集合中的元素"""try:values = self.redis.smembers(name)logger.info(f"SMEMBERS from set '{name}' with values {values}")return valuesexcept Exception as e:logger.error(f"Error getting members from set '{name}': {e}")return Nonedef close(self):"""关闭 Redis 连接"""try:self.redis.close()logger.info("Redis connection closed.")except Exception as e:logger.error(f"Error closing Redis connection: {e}")

2、redis-py-cluster 是一个 Python 库，它允许你使用 Redis 集群

• 安装redis-py-cluster

pip install redis-py-cluster

from rediscluster import RedisCluster
import logging# 设置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class RedisClusterUtils:def __init__(self, startup_nodes, decode_responses=True,password=None):"""初始化 Redis 集群连接:param startup_nodes: Redis 集群的初始节点列表:param decode_responses: 如果为 True，自动解码 Redis 返回的字节串为字符串"""self.startup_nodes = startup_nodesself.decode_responses = decode_responsesself.password = passwordtry:# 初始化 Redis 集群客户端self.redis = RedisCluster(startup_nodes=startup_nodes, decode_responses=decode_responses,password=password)logger.info("Connected to Redis cluster successfully.")except Exception as e:logger.error(f"Error connecting to Redis cluster: {e}")raisedef set(self, key, value, ex=None, px=None, nx=False, xx=False):"""设置 Redis 键值:param key: 键:param value: 值:param ex: 设置过期时间（秒）:param px: 设置过期时间（毫秒）:param nx: 只在键不存在时设置:param xx: 只在键存在时设置:return: 执行结果"""try:result = self.redis.set(key, value, ex=ex, px=px, nx=nx, xx=xx)logger.info(f"Set key '{key}' with value '{value}'")return resultexcept Exception as e:logger.error(f"Error setting key '{key}': {e}")return Nonedef get(self, key):"""获取 Redis 键的值:param key: 键:return: 值"""try:value = self.redis.get(key)logger.info(f"Get key '{key}' with value '{value}'")return valueexcept Exception as e:logger.error(f"Error getting key '{key}': {e}")return Nonedef delete(self, key):"""删除 Redis 键:param key: 键:return: 删除操作的结果"""try:result = self.redis.delete(key)logger.info(f"Deleted key '{key}'")return resultexcept Exception as e:logger.error(f"Error deleting key '{key}': {e}")return Nonedef hset(self, name, key, value):"""设置哈希表字段:param name: 哈希表名:param key: 字段:param value: 值:return: 执行结果"""try:result = self.redis.hset(name, key, value)logger.info(f"Set hash '{name}' field '{key}' with value '{value}'")return resultexcept Exception as e:logger.error(f"Error setting hash '{name}' field '{key}': {e}")return Nonedef hget(self, name, key):"""获取哈希表字段的值:param name: 哈希表名:param key: 字段:return: 字段值"""try:value = self.redis.hget(name, key)logger.info(f"Get hash '{name}' field '{key}' with value '{value}'")return valueexcept Exception as e:logger.error(f"Error getting hash '{name}' field '{key}': {e}")return Nonedef lpush(self, name, *values):"""向列表左侧插入数据:param name: 列表名:param values: 要插入的值:return: 执行结果"""try:result = self.redis.lpush(name, *values)logger.info(f"LPush to list '{name}' values {values}")return resultexcept Exception as e:logger.error(f"Error pushing to list '{name}': {e}")return Nonedef lrange(self, name, start, end):"""获取列表中的元素:param name: 列表名:param start: 起始索引:param end: 结束索引:return: 列表元素"""try:values = self.redis.lrange(name, start, end)logger.info(f"LRANGE from list '{name}' from index {start} to {end} with values {values}")return valuesexcept Exception as e:logger.error(f"Error getting range from list '{name}': {e}")return Nonedef pipeline(self):"""创建 Redis 管道操作:return: 管道对象"""return self.redis.pipeline()def scan(self, cursor=0, match=None, count=None):"""使用 SCAN 命令遍历 Redis 键:param cursor: 游标:param match: 匹配模式:param count: 每次扫描的数量:return: (游标, 键列表)"""try:cursor, keys = self.redis.scan(cursor, match=match, count=count)logger.info(f"SCAN with cursor {cursor}, found keys {keys}")return cursor, keysexcept Exception as e:logger.error(f"Error scanning keys: {e}")return cursor, []def close(self):"""关闭 Redis 连接"""try:self.redis.close()logger.info("Redis connection closed.")except Exception as e:logger.error(f"Error closing Redis connection: {e}")