.config/coc/extensions/node_modules/coc-pyright/node_modules/pyright/dist/typeshed-fallback/stubs/redis/redis/sentinel.pyi

   1 from typing import Any, Type, TypeVar, overload
   2
   3 from redis.client import Redis
   4 from redis.connection import Connection, ConnectionPool
   5 from redis.exceptions import ConnectionError
   6
   7 _Redis = TypeVar("_Redis", bound=Redis[Any])
   8
   9 class MasterNotFoundError(ConnectionError): ...
  10 class SlaveNotFoundError(ConnectionError): ...
  11
  12 class SentinelManagedConnection(Connection):
  13     connection_pool: Any
  14     def __init__(self, **kwargs) -> None: ...
  15     def connect_to(self, address) -> None: ...
  16     def connect(self) -> None: ...
  17     def read_response(self): ...
  18
  19 class SentinelConnectionPool(ConnectionPool):
  20     is_master: bool
  21     check_connection: bool
  22     connection_kwargs: Any
  23     service_name: str
  24     sentinel_manager: Any
  25     def __init__(self, service_name, sentinel_manager, **kwargs) -> None: ...
  26     def reset(self) -> None: ...
  27     def owns_connection(self, connection) -> bool: ...
  28     def get_master_address(self): ...
  29     def rotate_slaves(self): ...
  30
  31 # TODO: this should subclass `redis.commands.SentinelCommands` in the future
  32 # right now `redis.commands` is missing.
  33 class Sentinel(object):
  34     sentinel_kwargs: Any
  35     sentinels: Any
  36     min_other_sentinels: int
  37     connection_kwargs: Any
  38     def __init__(
  39         self, sentinels, min_other_sentinels: int = ..., sentinel_kwargs: Any | None = ..., **connection_kwargs
  40     ) -> None: ...
  41     def check_master_state(self, state, service_name) -> bool: ...
  42     def discover_master(self, service_name): ...
  43     def filter_slaves(self, slaves): ...
  44     def discover_slaves(self, service_name): ...
  45     @overload
  46     def master_for(self, service_name: str, *, connection_pool_class=..., **kwargs) -> Redis[Any]: ...
  47     @overload
  48     def master_for(self, service_name: str, redis_class: Type[_Redis] = ..., connection_pool_class=..., **kwargs) -> _Redis: ...
  49     @overload
  50     def slave_for(self, service_name: str, connection_pool_class=..., **kwargs) -> Redis[Any]: ...
  51     @overload
  52     def slave_for(self, service_name: str, redis_class: Type[_Redis] = ..., connection_pool_class=..., **kwargs) -> _Redis: ...