top_automatic_traceability/common/redis.py

import redis

from common.odps_data import OdpsDataCount


class SyncRedisHelper:
    _pool: redis.ConnectionPool = None
    _instance = None

    def __init__(self):
        if not self._instance:
            self._pool = self._get_pool()
            self._instance = self

    def _get_pool(self) -> redis.ConnectionPool:
        if self._pool is None:
            self._pool = redis.ConnectionPool(
                host="r-bp1mb0v08fqi4hjffupd.redis.rds.aliyuncs.com",  # 外网地址
                # host="r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com",  # 内网地址
                port=6379,
                db=0,
                password="Wqsd@2019",
                # password="Qingqu2019",

            )
        return self._pool

    def get_client(self) -> redis.Redis:
        pool = self._get_pool()
        client = redis.Redis(connection_pool=pool)
        return client

    def close(self):
        if self._pool:
            self._pool.disconnect(inuse_connections=True)

def insert_job_data(REDIS_NAME):
    data = OdpsDataCount.main()
    if not data:
        return 0
    helper = SyncRedisHelper()
    client = helper.get_client()
    client.rpush(REDIS_NAME, *data)
    return len(data)

def get_top_data(REDIS_NAME):
    helper = SyncRedisHelper()
    client = helper.get_client()
    ret = client.lpop(REDIS_NAME)
    return ret


def get_llen_list(REDIS_NAME):
    helper = SyncRedisHelper()
    client = helper.get_client()
    length = client.llen(REDIS_NAME)
    return length

def get_lrange_list(REDIS_NAME,count):
    helper = SyncRedisHelper()
    client = helper.get_client()
    elements = client.lrange(REDIS_NAME, 0, count)
    client.ltrim(REDIS_NAME, count, -1)
    return elements


def in_job_video_data(REDIS_NAME, ret):
    helper = SyncRedisHelper()
    client = helper.get_client()
    client.rpush(REDIS_NAME, str(ret))