import traceback import time import redis # import psycopg2 import pymysql # from psycopg2 import pool as pgpool from config import set_config from log import Log config_ = set_config() log_ = Log() conn_redis = None class RedisHelper(object): def __init__(self, params=None): """ 初始化redis连接信息 redis_info: redis连接信息, 格式：dict, {'host': '', 'port': '', 'password': ''} """ redis_info = config_.REDIS_INFO self.host = redis_info['host'] self.port = redis_info['port'] self.password = redis_info['password'] self.params = params def connect(self): """ 连接redis :return: conn """ global conn_redis if conn_redis is None: pool = redis.ConnectionPool(host=self.host, port=self.port, password=self.password, decode_responses=True) conn = redis.Redis(connection_pool=pool) conn_redis = conn return conn_redis def key_exists(self, key_name): """ 判断key是否存在 :param key_name: key :return: 存在-True, 不存在-False """ # start_time = time.time() conn = self.connect() res = conn.exists(key_name) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'get_data_from_redis', # 'executeTime': (time.time() - start_time) * 1000 # }) return res def del_keys(self, key_name): """ 删除key :param key_name: key :return: None """ conn = self.connect() conn.delete(key_name) def get_data_from_redis(self, key_name): """ 读取redis中的数据 :param key_name: key :return: data """ # start_time = time.time() conn = self.connect() if not conn.exists(key_name): # key不存在 return None data = conn.get(key_name) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'get_data_from_redis', # 'executeTime': (time.time() - start_time) * 1000 # }) return data def set_data_to_redis(self, key_name, value, expire_time=24*3600): """ 新增数据 :param key_name: key :param value: 元素的值 videoId :param expire_time: 过期时间，单位：s，默认1天 :return: None """ # start_time = time.time() conn = self.connect() conn.set(key_name, value, ex=int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'set_data_to_redis', # 'executeTime': (time.time() - start_time) * 1000}) def add_data_with_zset(self, key_name, data, expire_time=7*24*3600): """ 新增数据，有序set :param key_name: key :param data: 元素的值及对应分数 type-dict {value: score} :param expire_time: 过期时间，单位：s，默认7天，type-int :return: None """ # start_time = time.time() conn = self.connect() conn.zadd(key_name, data) # 设置过期时间 conn.expire(key_name, int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'add_data_with_zset', # 'executeTime': (time.time() - start_time) * 1000 # }) def get_data_zset_with_index(self, key_name, start, end, desc=True, with_scores=False): """ 根据索引位置获取元素的值 :param key_name: key :param start: 索引起始点 闭区间，包含start :param end: 索引结束点 闭区间，包含end :param desc: 分数排序方式，默认从大到小 :param with_scores: 是否获取元素的分数，默认 False，只获取元素的值 :return: data 元素值列表（不包含分数）,value(videoId)类型转换为int, 包含分数时不进行类型转换 """ # start_time = time.time() conn = self.connect() if not conn.exists(key_name): return None data = conn.zrange(key_name, start, end, desc, with_scores) if with_scores: data = data else: data = [eval(value) for value in data] # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'get_data_zset_with_index', # 'executeTime': (time.time() - start_time) * 1000 # }) return data def get_all_data_from_zset(self, key_name, desc=True, with_scores=False): """ 获取zset中所有元素的值 :param key_name: key :param desc: 分数排序方式，默认从大到小 :param with_scores: 是否获取元素的分数，默认 False，只获取元素的值 :return: data 元素值列表（不包含分数）,value(videoId)类型转换为int, 包含分数时不进行类型转换 """ conn = self.connect() if not conn.exists(key_name): return None data = [] start = 0 step = 100 while True: end = start + step - 1 temp = conn.zrange(key_name, start, end, desc, with_scores) if not temp: break data.extend(temp) start += step return data def get_score_with_value(self, key_name, value): """ 在zset中，根据元素的value获取对应的score :param key_name: key :param value: 元素的值 :return: score value对应的score """ conn = self.connect() if not conn.exists(key_name): return None return conn.zscore(key_name, value) def get_rank_with_value(self, key_name, value, desc=False): """ 在zset中，根据元素的value获取对应排名 :param key_name: key :param value: 元素的值 :param desc: 是否倒序 type-bool 默认：False-按照score从小到大 :return: rank value对应的rank，从0开始，不存在返回None """ conn = self.connect() if not conn.exists(key_name): return None if desc is True: return conn.zrevrank(key_name, value) else: return conn.zrank(key_name, value) def update_score_with_value(self, key_name, value, score, expire_time=24*3600): """ 在zset中，修改元素value对应的score :param key_name: key :param value: 元素的值 :param score: value对应的score更新值 :param expire_time: 过期时间，单位：s，默认1天，type-int """ conn = self.connect() if conn.exists(key_name): conn.zadd(key_name, {value: score}) else: # key不存在时，需设置过期时间 conn.zadd(key_name, {value: score}) conn.expire(key_name, int(expire_time)) def remove_value_from_zset(self, key_name, value): """ 删除zset中的指定元素 :param key_name: key :param value: 元素的值 :return: None """ # start_time = time.time() conn = self.connect() conn.zrem(key_name, value) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'remove_value_from_zset', # 'executeTime': (time.time() - start_time) * 1000 # }) def get_index_with_data(self, key_name, value): """ 根据元素的值获取在有序set中的位置，按照分数倒序（从大到小） :param key_name: key :param value: 元素的值 :return: idx 位置索引 """ # start_time = time.time() conn = self.connect() res = conn.zrevrank(key_name, value) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'get_index_with_data', # 'executeTime': (time.time() - start_time) * 1000 # }) return res def get_data_from_set(self, key_name): """ 获取set中的所有数据 :param key_name: key :return: data """ # start_time = time.time() conn = self.connect() if not conn.exists(key_name): # key不存在 return None data = [] cursor = 0 while True: cur, temp = conn.sscan(key_name, cursor=cursor, count=2000) data.extend(temp) if cur == 0: break cursor = cur # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'get_data_from_set', # 'executeTime': (time.time() - start_time) * 1000 # }) return list(set(data)) def add_data_with_set(self, key_name, values, expire_time=30*60): """ 新增数据，set :param key_name: key :param values: 要添加的元素 类型-tuple :param expire_time: 过期时间，单位：s，默认0.5小时 type-int :return: None """ # start_time = time.time() conn = self.connect() conn.sadd(key_name, *values) # 设置过期时间 conn.expire(key_name, int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'add_data_with_set', # 'executeTime': (time.time() - start_time) * 1000 # }) def data_exists_with_set(self, key_name, value): """ 判断元素value是否在集合key_name中 :param key_name: key :param value: 需判断的元素 :return: 存在-True, 不存在-False """ # start_time = time.time() conn = self.connect() res = conn.sismember(key_name, value) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'data_exists_with_set', # 'executeTime': (time.time() - start_time) * 1000 # }) return res def remove_value_from_set(self, key_name, values): """ 删除set中的指定元素 :param key_name: key :param values: 元素的值, 类型-tuple :return: None """ # start_time = time.time() conn = self.connect() conn.srem(key_name, *values) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'remove_value_from_set', # 'executeTime': (time.time() - start_time) * 1000 # }) def decr_key(self, key_name, amount=1, expire_time=30*60): """ redis自减 :param key_name: key :param amount: 自减数，默认为1，type-int :param expire_time: 过期时间，单位：s，默认0.5小时 type-int :return: None """ # start_time = time.time() conn = self.connect() conn.decr(name=key_name, amount=amount) conn.expire(key_name, int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'decr_key', # 'executeTime': (time.time() - start_time) * 1000 # }) def incr_key(self, key_name, amount=1, expire_time=30*60): """ redis自增 :param key_name: key :param amount: 自增数，默认为1，type-int :param expire_time: 过期时间，单位：s，默认0.5小时 type-int :return: None """ # start_time = time.time() conn = self.connect() conn.incr(name=key_name, amount=amount) conn.expire(key_name, int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'incr_key', # 'executeTime': (time.time() - start_time) * 1000 # }) def setnx_key(self, key_name, value, expire_time=5*60): """ 当key不存在时，将value塞入key中，key存在时不做操作 :param key_name: key :param value: value :return: 过期时间，单位：s，默认5分钟 type-int """ # start_time = time.time() conn = self.connect() conn.setnx(name=key_name, value=value) conn.expire(name=key_name, time=int(expire_time)) # if self.params is not None: # log_.info({ # 'logTimestamp': int(time.time() * 1000), # 'request_id': self.params.request_id, # 'operation': 'setnx_key', # 'executeTime': (time.time() - start_time) * 1000 # }) #hologres_info = config_.HOLOGRES_INFO #conn = psycopg2.connect(**hologres_info) #cur = conn.cursor() # holo连接 # connectPool = pgpool.SimpleConnectionPool(1, 30, **config_.HOLOGRES_INFO) # class HologresHelper(object): # def __init__(self): # """初始化hologres连接信息""" # #self.hologres_info = config_.HOLOGRES_INFO # # def get_data(self, sql): # #global conn # # 连接Hologres # #conn = psycopg2.connect(**self.hologres_info) # conn = connectPool.getconn() # # 创建游标 # cur = conn.cursor() # try: # # 查询数据 # cur.execute(sql) # data = cur.fetchall() # cur.close() # connectPool.putconn(conn, close=False) # # 提交事务 # #conn.commit() # # 释放资源 # #cur.close() # #conn.close() # except Exception as e: # log_.error('hologress error...') # log_.error(traceback.format_exc()) # cur.close() # connectPool.putconn(conn, close=False) # return [] # return data class MysqlHelper(object): def __init__(self): """ 初始化mysql连接信息 """ self.mysql_info = config_.MYSQL_INFO def get_data(self, sql): """ 查询数据 :param sql: sql语句 :return: data """ # 连接数据库 conn = pymysql.connect(**self.mysql_info) # 创建游标 cursor = conn.cursor() try: # 执行SQL语句 cursor.execute(sql) # 获取查询的所有记录 data = cursor.fetchall() except Exception as e: return None # 关闭游标对象 cursor.close() # 关闭数据库连接 conn.close() return data if __name__ == '__main__': redis_helper = RedisHelper() # key = 'flow.video.12345.123#112' # redis_helper.decr_key(key_name=key) # res = redis_helper.get_data_from_redis(key_name=key) # print(int(res), type(int(res))) # data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1) # print(len(data)) # key_name = 'com.weiqu.video.hot.recommend.previewed.4.weixin_openid_otjoB5VG780SB4aVjYqBBNLb - X6M' # values = (6134455, 9772930, 9912678, 9901969, 9926876, 9904203, 2384831, 9932272, 9737653, 9925240) key_name = 'com.weiqu.video.hot.recommend.previewed.4.abcd1' # values = (9902612, 9905573, 9928264, 9932148, 9809440, 9919900, 6093379, 9917093, 9793537, 9814345) # redis_helper.add_data_with_set(key_name=key_name, values=values, expire_time=30 * 60) res = redis_helper.get_data_from_set(key_name=key_name) print(res)