报告格式_天津市网站制作建设推广公司_传统营销方式有哪些_最新军事动态最新消息

在 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, Anyclass 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 Trueexcept Exception as e:print(f"Error setting key {key}: {e}")return Falsedef 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 Nonedef delete(self, key: str) -> bool:"""删除指定键:param key: 键:return: 是否成功"""try:self.redis_client.delete(key)return Trueexcept Exception as e:print(f"Error deleting key {key}: {e}")return Falsedef 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 Falsedef 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 Trueexcept Exception as e:print(f"Error setting multiple keys: {e}")return Falsedef 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 Nonedef 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 Nonedef 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 Trueexcept Exception as e:print(f"Error setting hash {hash_name}:{key}: {e}")return Falsedef 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 Nonedef 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'))

代码说明

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

通过这种封装，代码中的 Redis 操作会更加简洁，且提供了统一的接口，方便日后的维护和扩展。