Skip to content

RedisSentinelConnectionState

faststream.redis.configs.state.RedisSentinelConnectionState #

RedisSentinelConnectionState(
    options: dict[str, Any] | None = None,
    *,
    sentinels: Sequence[tuple[str, int]],
    master_name: str,
    sentinel_kwargs: Mapping[str, Any] | None = None,
)

Bases: RedisConnectionState

Builds the client via Sentinel.master_for for HA / failover.

The underlying SentinelConnectionPool re-discovers the current master on every reconnect, so publishers and stream consumers fail over for free (both go through connection.client).

Source code in faststream/redis/configs/state.py 
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    options: dict[str, Any] | None = None,
    *,
    sentinels: Sequence[tuple[str, int]],
    master_name: str,
    sentinel_kwargs: Mapping[str, Any] | None = None,
) -> None:
    super().__init__(options)
    self._sentinels = list(sentinels)
    self._master_name = master_name
    self._sentinel_kwargs = sentinel_kwargs

client property #

client: ClientT

connect async #

connect() -> Redis[bytes]
Source code in faststream/redis/configs/state.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def connect(self) -> "Redis[bytes]":
    # ``host``/``port`` describe a single node and are meaningless for
    # Sentinel — the master address is discovered from the sentinels.
    connection_kwargs = {
        k: v for k, v in self._options.items() if k not in {"host", "port"}
    }
    connection_kwargs["lib_name"] = "faststream"
    connection_kwargs["lib_version"] = __version__

    manager = Sentinel(
        self._sentinels,
        sentinel_kwargs=dict(self._sentinel_kwargs)
        if self._sentinel_kwargs is not None
        else None,
        **connection_kwargs,
    )
    client: Redis[bytes] = manager.master_for(self._master_name)

    self._client = client
    self._connected = True

    return client

disconnect async #

disconnect() -> None
Source code in faststream/redis/configs/state.py 
48
49
50
51
52
53
async def disconnect(self) -> None:
    if self._client:
        await self._client.aclose()  # type: ignore[attr-defined]

    self._client = None
    self._connected = False