Improve redis.sentinel types (#8331)

This commit is contained in:
Nikita Sobolev
2022-07-19 14:04:09 +03:00
committed by GitHub
parent e3d4bdc91a
commit 9cda16f5df

View File

@@ -1,5 +1,6 @@
from collections.abc import Iterable, Iterator
from typing import Any, TypeVar, overload
from typing_extensions import Literal
from typing_extensions import Literal, TypeAlias
from redis.client import Redis
from redis.commands.sentinel import SentinelCommands
@@ -7,16 +8,19 @@ 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: ...
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): ...
@@ -25,25 +29,29 @@ class SentinelConnectionPool(ConnectionPool):
check_connection: bool
connection_kwargs: Any
service_name: str
sentinel_manager: Any
def __init__(self, service_name, sentinel_manager, **kwargs) -> None: ...
sentinel_manager: Sentinel
def __init__(self, service_name: str, sentinel_manager: Sentinel, **kwargs) -> None: ...
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
self,
sentinels: Iterable[_AddressAndPort],
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