python Redis 操作工具类封装
在 Python 中封装 Redis 操作工具类，可以方便地在应用程序中复用 Redis 操作，简化 Redis 操作的复杂度，并提供统一的接口。这里给出一个基本的 Redis 操作工具类封装，使用 redis-py 库，这个库是 Python 与 Redis 交互的标准库。
1. 安装 redis-py
首先需要安装 redis-py 库。如果你没有安装，可以通过以下命令安装：
pip install redis
2. 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', 'Alice')
# 获取单个键值
print(redis_helper.get('name'))
# 设置多个键值对
redis_helper.set_multiple({'age': 30, 'location': 'New York'})
# 获取多个键值
print(redis_helper.get_multiple(['name', 'age', 'location']))
# 增加某个键的值
redis_helper.increment('age')
print(redis_helper.get('age'))
# 设置哈希表
redis_helper.hset('user:1000', 'name', 'Alice')
print(redis_helper.hget('user:1000', 'name'))
代码说明
- 初始化连接：在
__init__中，我们通过redis.StrictRedis初始化 Redis 连接，可以根据需要传递不同的 Redis 服务器信息。 - 基本操作封装：
set和get分别用于设置和获取键值。delete用于删除指定的键。exists用于检查键是否存在。
- 批量操作：提供了
set_multiple和get_multiple方法来处理批量的 Redis 操作。 - 自增和自减：封装了
increment和decrement方法来处理数字类型的键值的增减。 - 哈希表操作：通过
hset,hget, 和hgetall封装了对 Redis 哈希表的操作。
通过这种封装，代码中的 Redis 操作会更加简洁，且提供了统一的接口，方便日后的维护和扩展。