RODY/app/utils/redis_config.py

"""
redis 配置类
"""
import sys

import redis

from configs import default


class RedisCli(object):

    def __init__(self, *, host: str, port: str, password: str, db: int, socket_timeout=5):
        # redis对象，在 @app.on_event('startup') 中连接创建
        self._redis_client = None
        self.host = host
        self.port = port
        self.password = password
        self.db = db
        self.socket_timeout = socket_timeout

    def init_redis_connect(self) -> None:
        """
        初始化链接
        :return:
        """
        try:
            r = redis.ConnectionPool(
                host=self.host,
                port=self.port,
                password=self.password,
                db=self.db,
                socket_timeout=self.socket_timeout,
                decode_responses=True  # 解码
            )
            self._redis_client = redis.StrictRedis(connection_pool=r)
            if not self._redis_client.ping():
                print('连接超时')
                sys.exit()
        except (redis.AuthenticationError, Exception) as e:
            print('连接redis异常')
            sys.exit()

    def get_redis(self):
        return self._redis_client

    # 使实例化后的对象，赋予redis对象的方法和属性
    def __getattr__(self, item):
        return self._redis_client.has_key(item)

    def __getitem__(self, item):
        return self._redis_client[item]

    def __setitem__(self, key, value):
        self._redis_client[key] = value

    def __delitem__(self, key):
        del self._redis_client[key]


redis_client = RedisCli(
    host=default.db['host'],
    port=default.db['port'],
    password=default.db['password'],
    db=default.db['db'],
)