1 from typing import Any, Type, TypeVar, overload
3 from redis.client import Redis
4 from redis.connection import Connection, ConnectionPool
5 from redis.exceptions import ConnectionError
7 _Redis = TypeVar("_Redis", bound=Redis[Any])
9 class MasterNotFoundError(ConnectionError): ...
10 class SlaveNotFoundError(ConnectionError): ...
12 class SentinelManagedConnection(Connection):
14 def __init__(self, **kwargs) -> None: ...
15 def connect_to(self, address) -> None: ...
16 def connect(self) -> None: ...
17 def read_response(self): ...
19 class SentinelConnectionPool(ConnectionPool):
21 check_connection: bool
22 connection_kwargs: 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): ...
31 # TODO: this should subclass `redis.commands.SentinelCommands` in the future
32 # right now `redis.commands` is missing.
33 class Sentinel(object):
36 min_other_sentinels: int
37 connection_kwargs: Any
39 self, sentinels, min_other_sentinels: int = ..., sentinel_kwargs: Any | None = ..., **connection_kwargs
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): ...
46 def master_for(self, service_name: str, *, connection_pool_class=..., **kwargs) -> Redis[Any]: ...
48 def master_for(self, service_name: str, redis_class: Type[_Redis] = ..., connection_pool_class=..., **kwargs) -> _Redis: ...
50 def slave_for(self, service_name: str, connection_pool_class=..., **kwargs) -> Redis[Any]: ...
52 def slave_for(self, service_name: str, redis_class: Type[_Redis] = ..., connection_pool_class=..., **kwargs) -> _Redis: ...