[docs]classRedisSentinelStorage(RedisStorage):""" Rate limit storage with redis sentinel as backend Depends on :pypi:`redis` package """STORAGE_SCHEME=["redis+sentinel"]"""The storage scheme for redis accessed via a redis sentinel installation"""DEPENDENCIES={"redis.sentinel":Version("3.0")}def__init__(self,uri:str,service_name:Optional[str]=None,use_replicas:bool=True,sentinel_kwargs:Optional[Dict[str,Union[float,str,bool]]]=None,wrap_exceptions:bool=False,**options:Union[float,str,bool],)->None:""" :param uri: url of the form ``redis+sentinel://host:port,host:port/service_name`` :param service_name: sentinel service name (if not provided in :attr:`uri`) :param use_replicas: Whether to use replicas for read only operations :param sentinel_kwargs: kwargs to pass as :attr:`sentinel_kwargs` to :class:`redis.sentinel.Sentinel` :param wrap_exceptions: Whether to wrap storage exceptions in :exc:`limits.errors.StorageError` before raising it. :param options: all remaining keyword arguments are passed directly to the constructor of :class:`redis.sentinel.Sentinel` :raise ConfigurationError: when the redis library is not available or if the redis master host cannot be pinged. """super(RedisStorage,self).__init__(uri,wrap_exceptions=wrap_exceptions,**options)parsed=urllib.parse.urlparse(uri)sentinel_configuration=[]sentinel_options=sentinel_kwargs.copy()ifsentinel_kwargselse{}parsed_auth:Dict[str,Union[float,str,bool]]={}ifparsed.username:parsed_auth["username"]=parsed.usernameifparsed.password:parsed_auth["password"]=parsed.passwordsep=parsed.netloc.find("@")+1forlocinparsed.netloc[sep:].split(","):host,port=loc.split(":")sentinel_configuration.append((host,int(port)))self.service_name=(parsed.path.replace("/","")ifparsed.pathelseservice_name)ifself.service_nameisNone:raiseConfigurationError("'service_name' not provided")sentinel_dep=self.dependencies["redis.sentinel"].moduleself.sentinel:"redis.sentinel.Sentinel"=sentinel_dep.Sentinel(sentinel_configuration,sentinel_kwargs={**parsed_auth,**sentinel_options},**{**parsed_auth,**options},)self.storage=self.sentinel.master_for(self.service_name)self.storage_slave=self.sentinel.slave_for(self.service_name)self.use_replicas=use_replicasself.initialize_storage(uri)
[docs]defget(self,key:str)->int:""" :param key: the key to get the counter value for """returnsuper()._get(key,self.storage_slaveifself.use_replicaselseself.storage)
[docs]defget_expiry(self,key:str)->int:""" :param key: the key to get the expiry for """returnsuper()._get_expiry(key,self.storage_slaveifself.use_replicaselseself.storage)
[docs]defcheck(self)->bool:""" Check if storage is healthy by calling :class:`aredis.StrictRedis.ping` on the slave. """returnsuper()._check(self.storage_slaveifself.use_replicaselseself.storage)