BrandCultivation/database/db/redis_db.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import redis
from config import load_config

cfgs = load_config()


class RedisDatabaseHelper:
    _instance = None
    
    def __new__(cls):
        if not cls._instance:
            cls._instance = super(RedisDatabaseHelper, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance
        
    def __init__(self):
        if self._initialized:
            return
        self.redis = redis.StrictRedis(host=cfgs['redis']['host'],
                                       port=cfgs['redis']['port'],
                                       password=cfgs['redis']['passwd'],
                                       db=cfgs['redis']['db'],
                                       decode_responses=True)
        
        self._initialized = True


if __name__ == '__main__':
    import random
    # 连接到 Redis 服务器
    r = RedisDatabaseHelper().redis

    # 有序集合的键名
    zset_key = 'configs:hotkeys'

    data_list = ['ORDER_FULLORDR_RATE', 'MONTH6_SALE_QTY_YOY', 'MONTH6_SALE_QTY_MOM', 'MONTH6_SALE_QTY']

    # 清空已有的有序集合（可选，若需要全新的集合可执行此操作）
    r.delete(zset_key)
    
    for item in data_list:
        # 生成 80 到 100 之间的随机数，小数点后保留 4 位
        score = round(random.uniform(80, 100), 4)
        # 将元素和对应的分数添加到有序集合中
        r.zadd(zset_key, {item: score})

    # # 从 Redis 中读取有序集合并打印
    # result = r.zrange(zset_key, 0, -1, withscores=True)
    # for item, score in result:
    #     print(f"元素: {item}, 分数: {score}")