diff --git a/stubs/redis/redis/sentinel.pyi b/stubs/redis/redis/sentinel.pyi new file mode 100644 index 000000000..5ca32d666 --- /dev/null +++ b/stubs/redis/redis/sentinel.pyi @@ -0,0 +1,43 @@ +from typing import Any + +from redis.connection import Connection, ConnectionPool +from redis.exceptions import ConnectionError + +class MasterNotFoundError(ConnectionError): ... +class SlaveNotFoundError(ConnectionError): ... + +class SentinelManagedConnection(Connection): + connection_pool: Any + def __init__(self, **kwargs) -> None: ... + def connect_to(self, address) -> None: ... + def connect(self) -> None: ... + def read_response(self): ... + +class SentinelConnectionPool(ConnectionPool): + is_master: bool + check_connection: bool + connection_kwargs: Any + service_name: str + sentinel_manager: Any + def __init__(self, service_name, sentinel_manager, **kwargs) -> None: ... + def reset(self) -> None: ... + def owns_connection(self, connection) -> bool: ... + def get_master_address(self): ... + def rotate_slaves(self): ... + +# TODO: this should subclass `redis.commands.SentinelCommands` in the future +# right now `redis.commands` is missing. +class Sentinel(object): + sentinel_kwargs: Any + sentinels: Any + min_other_sentinels: int + connection_kwargs: Any + def __init__( + self, sentinels, min_other_sentinels: int = ..., sentinel_kwargs: Any | None = ..., **connection_kwargs + ) -> None: ... + def check_master_state(self, state, service_name) -> bool: ... + def discover_master(self, service_name): ... + def filter_slaves(self, slaves): ... + def discover_slaves(self, service_name): ... + def master_for(self, service_name, redis_class=..., connection_pool_class=..., **kwargs) -> None: ... + def slave_for(self, service_name, redis_class=..., connection_pool_class=..., **kwargs) -> None: ...