73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
|
import os
|
|||
|
|
import json
|
|||
|
|
import redis
|
|||
|
|
from typing import List, Dict, Any, Optional
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RedisTaskManager:
|
|||
|
|
def __init__(self, config_data: Dict[str, Any]):
|
|||
|
|
self.config_data = config_data
|
|||
|
|
"""初始化Redis连接"""
|
|||
|
|
self.redis_client = redis.Redis(
|
|||
|
|
host=self.config_data['redis_config']['host'],
|
|||
|
|
port=self.config_data['redis_config']['port'],
|
|||
|
|
password=self.config_data['redis_config']['password'],
|
|||
|
|
db=self.config_data['redis_config']['db'],
|
|||
|
|
decode_responses=True
|
|||
|
|
)
|
|||
|
|
self.expire_hours = 24 # 过期时间24小时
|
|||
|
|
|
|||
|
|
def write_data(self, key: str, data: Any, expire_time: Optional[int] = None):
|
|||
|
|
"""
|
|||
|
|
写入数据到Redis，设置过期时间
|
|||
|
|
Args:
|
|||
|
|
key: Redis键名
|
|||
|
|
data: 要存储的数据
|
|||
|
|
expire_time: 自定义过期时间（秒），默认24小时
|
|||
|
|
Returns:
|
|||
|
|
bool: 写入是否成功
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
# 如果数据是字典或列表，先序列化为JSON
|
|||
|
|
if isinstance(data, (dict, list)):
|
|||
|
|
data_str = json.dumps(data, ensure_ascii=False)
|
|||
|
|
else:
|
|||
|
|
data_str = str(data)
|
|||
|
|
# 设置过期时间，优先使用自定义时间
|
|||
|
|
expire_seconds = expire_time if expire_time is not None else self.expire_hours * 3600
|
|||
|
|
result = self.redis_client.setex(key, expire_seconds, data_str)
|
|||
|
|
|
|||
|
|
if result:
|
|||
|
|
print(f"数据写入成功 - 键: {key}, 过期时间: {expire_seconds}秒")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
print(f"数据写入失败 - 键: {key}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"写入Redis数据时发生错误: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def read_data(self, key: str):
|
|||
|
|
"""
|
|||
|
|
从Redis读取数据
|
|||
|
|
Args:
|
|||
|
|
key: Redis键名
|
|||
|
|
Returns:
|
|||
|
|
Optional[Any]: 读取到的数据，如果键不存在返回None
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
data_str = self.redis_client.get(key)
|
|||
|
|
if data_str is None:
|
|||
|
|
print(f"键不存在或已过期 - 键: {key}")
|
|||
|
|
return None
|
|||
|
|
# 尝试解析JSON数据
|
|||
|
|
try:
|
|||
|
|
return json.loads(data_str)
|
|||
|
|
except json.JSONDecodeError:
|
|||
|
|
# 如果不是JSON格式，返回原始字符串
|
|||
|
|
return data_str
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"读取Redis数据时发生错误: {e}")
|
|||
|
|
return None
|