base_redis_service.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@Email: ludq1@chinaunicom.cn
@date: 2023/04/07 11:40:00
@Description:
"""

import random
from typing import List, Dict

import redis
from redis import StrictRedis
from redis.sentinel import Sentinel

from .common_app_config import CommonAppConfig
from .utils_base import UtilityBaseV2


class BaseRedisService:
    r"""
    redis操作基类
    """
    # 当调用jedis  API时发生错误后重试的次数, 默认为1
    retry_times_when_exception: int = 1
    app_utils: UtilityBaseV2 = None
    need_decode_responses: bool = True
    r"""
    获取到redis函数结果后是否需要解码
    
    如果conn的初始化中包含 decode_responses=True，则应该设置此属性为False，
    
    即获取到的redis函数结果已经是解码后的数据了，不需要再次解码
    """
    encoding_used_for_decode_response: str = "utf-8"

    def __init__(self, *args, **kwargs):
        r"""
        初始化

        :key retry_times_when_exception 当操作发生异常时的重试次数，默认为 1
        :key app_utils: 使用的app_utils便捷类，默认自动生成
        :key logger: 用于打印日志
        """
        retry_times_when_exception = kwargs.get("retry_times_when_exception")
        app_utils = kwargs.get("app_utils")

        self.retry_times_when_exception = retry_times_when_exception if retry_times_when_exception > 0 else 1
        self.app_utils = app_utils if app_utils else UtilityBaseV2(
            logger=kwargs.get("logger") or CommonAppConfig().common_logger,
        )
        self.logger = kwargs.get("logger") or CommonAppConfig().common_logger

    def set(self, name: str, value: str, ex: int = None, px: int = None, nx: bool = False, xx: bool = False) -> bool:
        r"""
        Set the value at key ``name`` to ``value`` ， 返回 bool 表示是否成功执行

        :param name
        :param value
        :param ex sets an expire flag on key ``name`` for ``ex`` seconds.
        :param px sets an expire flag on key ``name`` for ``px`` milliseconds.
        :param nx if set to True, set the value at key ``name`` to ``value`` only if it does not exist.
        :param xx if set to True, set the value at key ``name`` to ``value`` only if it already exists.
        """

        def operation_func():
            return self.get_master_conn().set(name=name, value=value, ex=ex, px=px, nx=nx, xx=xx)

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def get(self, name: str) -> str:
        r"""
        Return the value at key ``name``, or None if the key doesn't exist,返回 utf-8解码的字符串

        :param name
        :return

        """

        def operation_func():
            result = self.get_slave_conn().get(name=name)
            if self.need_decode_responses and result is not None:
                result = str(result, self.encoding_used_for_decode_response)
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def mset(self, name_value_dict: dict) -> bool:
        r"""
        Sets key/values based on a mapping. Mapping is a dictionary of
        key/value pairs. Both keys and values should be strings or types that
        can be cast to a string via str() ，

        返回 bool 表示是否成功执行

        :param name_value_dict
        :return
        """

        def operation_func():
            return self.get_master_conn().mset(mapping=name_value_dict)

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def mget(self, *to_get_names: str) -> list:
        r"""
        Returns a list of values ordered identically to ``keys``,

        返回 utf-8解码的字符串list

        :param to_get_names
        :return
        """

        def operation_func():
            result = self.get_slave_conn().mget(keys=to_get_names)
            if self.need_decode_responses and result is not None:
                result = [
                    str(tmp_byte_value, self.encoding_used_for_decode_response) if tmp_byte_value is not None else None
                    for
                    tmp_byte_value in
                    result]
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def hset(self, name: str, key: str, value: str) -> int:
        r"""
        Set ``key`` to ``value`` within hash ``name``

        Returns 1 if HSET created a new field, otherwise 0， 返回 int 表示是否成功插入的key的数量

        :param name
        :param key
        :param value
        """

        def operation_func():
            return self.get_master_conn().hset(name=name, key=key, value=value)

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def hget(self, name: str, key: str) -> str:
        r"""
        Returns a list of values ordered identically to ``keys``,返回 utf-8解码的字符串list

        :param name
        :param key
        """

        def operation_func():
            result = self.get_slave_conn().hget(name=name, key=key)
            if self.need_decode_responses and result is not None:
                result = str(result, self.encoding_used_for_decode_response)
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def hmset(self, name: str, key_value_dict: dict) -> bool:
        r"""
        Set key to value within hash ``name`` for each corresponding
        key and value from the ``mapping`` dict. ，

        返回 bool 表示是否成功执行

        :param name
        :param key_value_dict
        """

        def operation_func():
            return self.get_master_conn().hmset(name=name, mapping=key_value_dict)

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def hmget(self, *to_get_keys: str, name: str) -> list:
        r"""
        Returns a list of values ordered identically to ``keys``,

        返回 utf-8解码的字符串list

        :param to_get_keys
        :param name
        """

        def operation_func():
            result = self.get_slave_conn().hmget(name=name, keys=to_get_keys)
            if self.need_decode_responses and result is not None:
                result = [
                    str(tmp_byte_value, self.encoding_used_for_decode_response) if tmp_byte_value is not None else None
                    for
                    tmp_byte_value in
                    result]
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def hgetall(self, name: str) -> dict:
        r"""
        Return a Python dict of the hash's name/value pairs ,

        返回 utf-8解码的key,value字符串 的 dict

        :param name
        """

        def operation_func():
            result = self.get_slave_conn().hgetall(name=name)
            if self.need_decode_responses and result is not None:
                result = {str(tmp_byte_key, self.encoding_used_for_decode_response): str(result[tmp_byte_key],
                                                                                         self.encoding_used_for_decode_response)
                          for
                          tmp_byte_key in
                          result.keys()}
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def expire(self, name: str, time: int, use_second: bool = False, use_millisecond: bool = False) -> bool:
        r"""
        设置key的过期时间, 从现在开始算过多少秒或毫秒之后key会被删除

        如果 use_second 为 True ,则 time 单位为秒的int或者 python timedelta object.

        如果 use_millisecond 为 True , 则 time 单位为毫秒的int或者 python timedelta object.

        use_second 和 use_millisecond 必须至少有一个为True

        返回 bool 表示是否成功执行

        :param name
        :param time
        :param use_millisecond
        :param use_second
        """

        def operation_func():
            if use_second:
                return self.get_master_conn().expire(name=name, time=time)
            elif use_millisecond:
                return self.get_master_conn().pexpire(name=name, time=time)
            else:
                raise RuntimeError('use_second 和 use_millisecond 必须至少有一个为True')

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def expireat(self, name: str, when: int, use_second: bool = False, use_millisecond: bool = False) -> bool:
        r"""
        设置key的过期时间, 在指定的时间点key会被删除

        如果 use_second 为 True ,则 when 为 unix time or a Python datetime object

        如果 use_millisecond 为 True , 则 when 为 unix time in milliseconds (unix time * 1000) or a Python datetime object.

        use_second 和 use_millisecond 必须至少有一个为True

        返回 bool 表示是否成功执行

        :param name
        :param when
        :param use_second
        :param use_millisecond
        """

        def operation_func():
            if use_second:
                return self.get_master_conn().expireat(name=name, when=when)
            elif use_millisecond:
                return self.get_master_conn().pexpireat(name=name, when=when)
            else:
                raise RuntimeError('use_second 和 use_millisecond 必须至少有一个为True')

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def delete(self, *to_delete_names: str) -> int:
        r"""
        Delete one or more keys specified by ``names``

        返回 int 表示是否成功删除的key的数量

        :param to_delete_names
        """

        def operation_func():
            return self.get_master_conn().delete(*to_delete_names)

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def keys(self, pattern: str) -> list:
        r"""
        Returns a list of keys matching ``pattern``,

        返回 utf-8解码的字符串list

        :param pattern
        """

        def operation_func():
            result = self.get_slave_conn().keys(pattern=pattern)
            if self.need_decode_responses and result is not None:
                result = [str(tmp_byte_value, self.encoding_used_for_decode_response) for tmp_byte_value in result]
            return result

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def delete_by_pattern(self, pattern: str) -> int:
        r"""
        按 pattern 删除key，

        返回 int 表示是否成功删除的key的数量

        :param pattern
        """

        def operation_func():
            key_list = self.get_slave_conn().keys(pattern=pattern)
            if key_list:
                return self.get_master_conn().delete(*key_list)
            else:
                return 0

        return self.app_utils.common_operation(retry_times_when_exception=self.retry_times_when_exception,
                                               operation_func=operation_func,
                                               on_operation_failed=self.on_operation_failed,
                                               on_operation_completed=self.on_operation_completed)

    def on_operation_failed(self):
        r"""
        当redis操作失败时的动作，一般是重建连接，或者等待几秒之类的操作

        :return:
        """
        pass

    def on_operation_completed(self):
        r"""
        当redis操作完成时的动作，一般是什么都不做

        :return:
        """
        pass

    def get_master_conn(self) -> StrictRedis:
        r"""
        获取master连接

        :return:
        """
        raise RuntimeError("Not implemented")

    def get_slave_conn(self) -> StrictRedis:
        r"""
        获取slave连接

        :return:
        """
        raise RuntimeError("Not implemented")


class RedisServiceMasterAndSlave(BaseRedisService):
    r"""
    master redis和slave redis分别设置的redis帮助服务
    """

    master_redis_config_dict_list: List[Dict] = None
    slave_redis_config_dict_list: List[Dict] = None
    master_redis_conn_list: List[StrictRedis] = None
    slave_redis_conn_list: List[StrictRedis] = None

    def __init__(self, *args, **kwargs):
        r"""
        初始化

        :key retry_times_when_exception 当操作发生异常时的重试次数，默认为 1
        :key app_utils: 使用的app_utils便捷类，默认自动生成
        :key master_redis_config_dict_list master redis连接的dict列表
        :key slave_redis_config_dict_list slave redis连接的dict列表
            字典的key和value组成参考 redis.Redis()的init函数,大致内容有
             host='localhost', port=6379,
             db=0, password=None, socket_timeout=None,
             socket_connect_timeout=None,
             socket_keepalive=None, socket_keepalive_options=None,
             connection_pool=None, unix_socket_path=None,
             encoding='utf-8', encoding_errors='strict',
             charset=None, errors=None,
             decode_responses=False, retry_on_timeout=False,
             ssl=False, ssl_keyfile=None, ssl_certfile=None,
             ssl_cert_reqs='required', ssl_ca_certs=None,
             ssl_check_hostname=False,
             max_connections=None, single_connection_client=False,
             health_check_interval=0, client_name=None, username=None
        """
        super().__init__(*args, **kwargs)

        self.master_redis_config_dict_list = kwargs.get("master_redis_config_dict_list")
        self.slave_redis_config_dict_list = kwargs.get("slave_redis_config_dict_list")

        self.need_decode_responses = False if self.master_redis_config_dict_list[0].get("decode_responses") else True
        encoding_used_for_decode_response = self.master_redis_config_dict_list[0].get("encoding")
        self.encoding_used_for_decode_response = encoding_used_for_decode_response if encoding_used_for_decode_response else 'utf-8'

        self._init_redis_conn()

    def get_master_conn(self):
        return self.master_redis_conn_list[random.choice(range(len(self.master_redis_conn_list)))]

    def get_slave_conn(self):
        return self.slave_redis_conn_list[random.choice(range(len(self.slave_redis_conn_list)))]

    def on_operation_failed(self):
        self._init_redis_conn()

    def _init_redis_conn(self):
        self.master_redis_conn_list = [redis.StrictRedis(**tmp_dict) for tmp_dict in self.master_redis_config_dict_list]
        self.slave_redis_conn_list = [redis.StrictRedis(**tmp_dict) for tmp_dict in self.slave_redis_config_dict_list]


class RedisServiceSentinel(BaseRedisService):
    r"""
    哨兵版redis集群的redis帮助服务
    """
    redis_sentinel_config_dict: dict = None
    sentinel: Sentinel = None

    service_name: str = None

    def __init__(self, *args, **kwargs):
        r"""
        初始化

        :key retry_times_when_exception 当操作发生异常时的重试次数，默认为 1
        :key app_utils: 使用的app_utils便捷类，默认自动生成
        :key service_name: 通过哨兵查询redis server使用的服务名
        :key redis_sentinel_config_dict   redis哨兵连接的dict配置信息
            字典的key和value组成参考 redis.Sentinel 的init函数,大致内容有
            sentinels 一组节点，每个节点使用 (hostname, port) 表示
            min_other_sentinels 为一个哨兵定义一个peers的最小节点数，当查询一个哨兵时，如果没有达到这个阈值，其响应不被认为是有效的,默认为0
            sentinel_kwargs 是一个字典,用于连接到哨兵时使用, 和 redis.Redis 的init函数中需要的参数一样，
            如果没有提供该值，则使用 connection_kwargs 中以 socket_ 开头的参数作为 sentinel_kwargs
            connection_kwargs 中的参数用于创建 redis server时使用
            基本上 sentinel_kwargs 只需要 password=None, socket_timeout=None,
            connection_kwargs 中需要 db=0, password=None, socket_timeout=None,
            其中 sentinel_kwargs 和 connection_kwargs 可用的参数有
             db=0, password=None, socket_timeout=None,
             socket_connect_timeout=None,
             socket_keepalive=None, socket_keepalive_options=None,
             connection_pool=None, unix_socket_path=None,
             encoding='utf-8', encoding_errors='strict',
             charset=None, errors=None,
             decode_responses=False, retry_on_timeout=False,
             ssl=False, ssl_keyfile=None, ssl_certfile=None,
             ssl_cert_reqs='required', ssl_ca_certs=None,
             ssl_check_hostname=False,
             max_connections=None, single_connection_client=False,
             health_check_interval=0, client_name=None, username=None
        """
        super().__init__(*args, **kwargs)

        self.service_name = kwargs.get("service_name")
        self.redis_sentinel_config_dict = kwargs.get("redis_sentinel_config_dict")

        sentinels = self.redis_sentinel_config_dict.get("sentinels")
        min_other_sentinels = self.redis_sentinel_config_dict.get("min_other_sentinels")
        sentinel_kwargs = self.redis_sentinel_config_dict.get("sentinel_kwargs")
        connection_kwargs = self.redis_sentinel_config_dict.get("connection_kwargs")

        self.need_decode_responses = False if connection_kwargs and connection_kwargs.get("decode_responses") else True
        encoding_used_for_decode_response = connection_kwargs.get("encoding") if connection_kwargs else None
        self.encoding_used_for_decode_response = encoding_used_for_decode_response if encoding_used_for_decode_response else 'utf-8'

        if connection_kwargs:
            self.sentinel = Sentinel(sentinels=sentinels, min_other_sentinels=min_other_sentinels,
                                     sentinel_kwargs=sentinel_kwargs, **connection_kwargs)
        else:
            self.sentinel = Sentinel(sentinels=sentinels, min_other_sentinels=min_other_sentinels,
                                     sentinel_kwargs=sentinel_kwargs)

    def get_master_conn(self):
        return self.sentinel.master_for(service_name=self.service_name)

    def get_slave_conn(self):
        return self.sentinel.slave_for(service_name=self.service_name)

    def on_operation_failed(self):
        pass


class RedisServiceCluster(BaseRedisService):
    r"""
    集群版redis集群的redis帮助服务
    """

    redis_server_config_dict_list: List[Dict] = None
    redis_server_conn_list: List[StrictRedis] = None

    def __init__(self, *args, **kwargs):
        r"""
        初始化

        :key retry_times_when_exception 当操作发生异常时的重试次数，默认为 1
        :key app_utils: 使用的app_utils便捷类，默认自动生成
        :key redis_server_config_dict_list   redis连接的dict列表
            字典的key和value组成参考 redis.Redis()的init函数,大致内容有
             host='localhost', port=6379,
             db=0, password=None, socket_timeout=None,
             socket_connect_timeout=None,
             socket_keepalive=None, socket_keepalive_options=None,
             connection_pool=None, unix_socket_path=None,
             encoding='utf-8', encoding_errors='strict',
             charset=None, errors=None,
             decode_responses=False, retry_on_timeout=False,
             ssl=False, ssl_keyfile=None, ssl_certfile=None,
             ssl_cert_reqs='required', ssl_ca_certs=None,
             ssl_check_hostname=False,
             max_connections=None, single_connection_client=False,
             health_check_interval=0, client_name=None, username=None
        """
        super().__init__(*args, **kwargs)

        self.redis_server_config_dict_list = kwargs.get("redis_server_config_dict_list")

        self.need_decode_responses = False if self.redis_server_config_dict_list[0].get("decode_responses") else True
        encoding_used_for_decode_response = self.redis_server_config_dict_list[0].get("encoding")
        self.encoding_used_for_decode_response = encoding_used_for_decode_response if encoding_used_for_decode_response else 'utf-8'

        self._init_redis_conn()

    def get_master_conn(self):
        return self.redis_server_conn_list[random.choice(range(len(self.redis_server_conn_list)))]

    def get_slave_conn(self):
        return self.redis_server_conn_list[random.choice(range(len(self.redis_server_conn_list)))]

    def on_operation_failed(self):
        self._init_redis_conn()

    def _init_redis_conn(self):
        self.redis_server_conn_list = [redis.StrictRedis(**tmp_dict) for tmp_dict in self.redis_server_config_dict_list]