Skip to content
Merged
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 18 additions & 14 deletions stubs/redis/redis/sentinel.pyi
Original file line number Diff line number Diff line change
@@ -1,22 +1,26 @@
from typing import Any, TypeVar, overload
from typing_extensions import Literal
from collections.abc import Iterator
from typing import Any, Iterable, TypeVar, overload
Comment thread
sobolevn marked this conversation as resolved.
Outdated
from typing_extensions import Literal, TypeAlias

from redis.client import Redis
from redis.commands.sentinel import SentinelCommands
from redis.connection import Connection, ConnectionPool, SSLConnection
from redis.exceptions import ConnectionError

_RedisT = TypeVar("_RedisT", bound=Redis[Any])
_AddressAndPort: TypeAlias = tuple[str, int]
_SentinelState: TypeAlias = dict[str, Any] # TODO: this can be a TypedDict

class MasterNotFoundError(ConnectionError): ...
class SlaveNotFoundError(ConnectionError): ...

class SentinelManagedConnection(Connection):
connection_pool: Any
connection_pool: SentinelConnectionPool
def __init__(self, **kwargs) -> None: ...
def connect_to(self, address) -> None: ...
def connect_to(self, address: _AddressAndPort) -> None: ...
Comment thread
srittau marked this conversation as resolved.
def connect(self) -> None: ...
def read_response(self, disable_decoding: bool = ...): ...
# The result can be either `str | bytes` or `list[str | bytes]`
def read_response(self, disable_decoding: bool = ...) -> Any: ...

class SentinelManagedSSLConnection(SentinelManagedConnection, SSLConnection): ...

Expand All @@ -25,25 +29,25 @@ class SentinelConnectionPool(ConnectionPool):
check_connection: bool
connection_kwargs: Any
service_name: str
sentinel_manager: Any
sentinel_manager: Sentinel
def __init__(self, service_name, sentinel_manager, **kwargs) -> None: ...
Comment thread
sobolevn marked this conversation as resolved.
Outdated
def reset(self) -> None: ...
def owns_connection(self, connection) -> bool: ...
def get_master_address(self): ...
def rotate_slaves(self): ...
def owns_connection(self, connection: Connection) -> bool: ...
def get_master_address(self) -> _AddressAndPort: ...
def rotate_slaves(self) -> Iterator[_AddressAndPort]: ...

class Sentinel(SentinelCommands):
sentinel_kwargs: Any
sentinels: Any
sentinels: list[Redis[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 check_master_state(self, state: _SentinelState, service_name: str) -> bool: ...
def discover_master(self, service_name: str) -> _AddressAndPort: ...
def filter_slaves(self, slaves: Iterable[_SentinelState]) -> list[_AddressAndPort]: ...
def discover_slaves(self, service_name: str) -> list[_AddressAndPort]: ...
@overload
def master_for(self, service_name: str, *, connection_pool_class=..., **kwargs) -> Redis[Any]: ...
@overload
Expand Down