rov-server/aiocache.py

import aioredis
import aiopg
from config import set_config
from log import Log

config_ = set_config()
log = Log()

redis_info = config_.REDIS_INFO

# R = get_redis_connect()

pg_info = config_.HOLOGRES_INFO
# P = aiopg.connect(database=pg_info['dbname'],
#                                user=pg_info['user'],
#                                password=pg_info['password'],
#                                host=pg_info['host'])


R = aioredis.from_url(redis_info['host'], password=redis_info['password'])
# async def get_redis_connect():
#     return await aioredis.from_url(redis_info['host'], password=redis_info['password'])

# async def pg_getdata(sql):
#     P = await aiopg.connect(database=pg_info['dbname'],
#                                user=pg_info['user'],
#                                password=pg_info['password'],
#                                host=pg_info['host'],
#                                port=pg_info['port'])
#     cur = await P.cursor()
#     await cur.execute(sql)
#     ret = await cur.fetchall()
#     return ret

dsn = f'dbname={pg_info["dbname"]} user={pg_info["user"]} password={pg_info["password"]} host={pg_info["host"]} port={pg_info["port"]}'
print(dsn)
async def pg_getdata(sql):
    pool = await aiopg.create_pool(dsn)
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(sql)    
            return await cur.fetchall()

async def key_exists(key_name):
    """
    判断key是否存在
    :param key_name: key
    :return: 存在-True, 不存在-False
    """
    return await R.exists(key_name)

async def del_keys(key_name):
    """
    删除key
    :param key_name: key
    :return: None
    """
    return await R.delete(key_name)

async def get_data_from_redis(key_name):
    """
    读取redis中的数据
    :param key_name: key
    :return: data
    """
    if not await R.exists(key_name):
        # key不存在
        return None
    data = await R.get(key_name)
    return data

async def set_data_to_redis(key_name, value, expire_time=24*3600):
    """
    新增数据
    :param key_name: key
    :param value: 元素的值 videoId
    :param expire_time: 过期时间，单位：s，默认1天
    :return: None
    """
    await R.set(key_name, value, ex=expire_time)

async def add_data_with_zset(key_name, data, expire_time=7*24*3600):
    """
    新增数据，有序set
    :param key_name: key
    :param data: 元素的值及对应分数 type-dict  {value: score}
    :param expire_time: 过期时间，单位：s，默认7天
    :return: None
    """
    await R.zadd(key_name, data)
    # 设置过期时间
    await R.expire(key_name, int(expire_time))

async def get_data_zset_with_index(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, 包含分数时不进行类型转换
    """
    if not await R.exists(key_name):
        return None
    data = await R.zrange(key_name, start, end, desc, with_scores)
    if with_scores:
        return data
    else:
        return [eval(value) for value in data]

async def get_score_with_value(key_name, value):
    """
    在zset中，根据元素的value获取对应的score
    :param key_name: key
    :param value: 元素的值
    :return: score value对应的score
    """
    if not await R.exists(key_name):
        return None
    return await R.zscore(key_name, value)

async def update_score_with_value(key_name, value, score, expire_time=7*24*3600):
    """
    在zset中，修改元素value对应的score
    :param key_name: key
    :param value: 元素的值
    :param score: value对应的score更新值
    :param expire_time: 过期时间，单位：s，默认7天
    """
    if R.exists(key_name):
        R.zadd(key_name, {value: score})
    else:
        # key不存在时，需设置过期时间
        R.zadd(key_name, {value: score})
        R.expire(key_name, expire_time)

async def remove_value_from_zset(key_name, value):
    """
    删除zset中的指定元素
    :param key_name: key
    :param value: 元素的值
    :return: None
    """
    R.zrem(key_name, value)

async def get_index_with_data(key_name, value):
    """
    根据元素的值获取在有序set中的位置，按照分数倒序（从大到小）
    :param key_name: key
    :param value: 元素的值
    :return: idx 位置索引
    """
    return await R.zrevrank(key_name, value)

async def get_data_from_set(key_name):
    """
    获取set中的所有数据
    :param key_name: key
    :return: data
    """
    if not await R.exists(key_name):
        # key不存在
        return None
    data = await R.sscan(key_name)
    return data[1]

async def add_data_with_set(key_name, values, expire_time=30*60):
    """
    新增数据，set
    :param key_name: key
    :param values: 要添加的元素  类型-tuple
    :param expire_time: 过期时间，单位：s，默认0.5小时
    :return: None
    """
    R.sadd(key_name, *values)
    # 设置过期时间
    R.expire(key_name, expire_time)

async def data_exists_with_set(key_name, value):
    """
    判断元素value是否在集合key_name中
    :param key_name: key
    :param value: 需判断的元素
    :return: 存在-True, 不存在-False
    """
    return await R.sismember(key_name, value)

async def remove_value_from_set(key_name, values):
    """
    删除set中的指定元素
    :param key_name: key
    :param values: 元素的值, 类型-tuple
    :return: None
    """
    await R.srem(key_name, *values)