Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import warnings
2from typing import Any, Awaitable, overload
4from redis.typing import (
5 AsyncClientProtocol,
6 SentinelMasterAddress,
7 SentinelMastersResponse,
8 SyncClientProtocol,
9)
12class SentinelCommands:
13 """
14 A class containing the commands specific to redis sentinel. This class is
15 to be used as a mixin.
16 """
18 def sentinel(self, *args):
19 """Redis Sentinel's SENTINEL command."""
20 warnings.warn(DeprecationWarning("Use the individual sentinel_* methods"))
22 @overload
23 def sentinel_get_master_addr_by_name(
24 self: SyncClientProtocol,
25 service_name,
26 return_responses: bool = False,
27 ) -> SentinelMasterAddress | bool: ...
29 @overload
30 def sentinel_get_master_addr_by_name(
31 self: AsyncClientProtocol,
32 service_name,
33 return_responses: bool = False,
34 ) -> Awaitable[SentinelMasterAddress | bool]: ...
36 def sentinel_get_master_addr_by_name(
37 self, service_name, return_responses: bool = False
38 ) -> (SentinelMasterAddress | bool) | Awaitable[SentinelMasterAddress | bool]:
39 """
40 Returns a (host, port) pair for the given ``service_name`` when return_responses is True,
41 otherwise returns a boolean value that indicates if the command was successful.
42 """
43 return self.execute_command(
44 "SENTINEL GET-MASTER-ADDR-BY-NAME",
45 service_name,
46 once=True,
47 return_responses=return_responses,
48 )
50 @overload
51 def sentinel_master(
52 self: SyncClientProtocol,
53 service_name,
54 return_responses: bool = False,
55 ) -> dict[str, Any] | bool: ...
57 @overload
58 def sentinel_master(
59 self: AsyncClientProtocol,
60 service_name,
61 return_responses: bool = False,
62 ) -> Awaitable[dict[str, Any] | bool]: ...
64 def sentinel_master(self, service_name, return_responses: bool = False) -> (
65 dict[str, Any] | bool
66 ) | Awaitable[dict[str, Any] | bool]:
67 """
68 Returns a dictionary containing the specified masters state, when return_responses is True,
69 otherwise returns a boolean value that indicates if the command was successful.
70 """
71 return self.execute_command(
72 "SENTINEL MASTER", service_name, return_responses=return_responses
73 )
75 @overload
76 def sentinel_masters(self: SyncClientProtocol) -> SentinelMastersResponse: ...
78 @overload
79 def sentinel_masters(
80 self: AsyncClientProtocol,
81 ) -> Awaitable[SentinelMastersResponse]: ...
83 def sentinel_masters(
84 self,
85 ) -> SentinelMastersResponse | Awaitable[SentinelMastersResponse]:
86 """
87 Returns a list of dictionaries containing each master's state.
89 Important: This function is called by the Sentinel implementation and is
90 called directly on the Redis standalone client for sentinels,
91 so it doesn't support the "once" and "return_responses" options.
92 """
93 return self.execute_command("SENTINEL MASTERS")
95 @overload
96 def sentinel_monitor(self: SyncClientProtocol, name, ip, port, quorum) -> bool: ...
98 @overload
99 def sentinel_monitor(
100 self: AsyncClientProtocol, name, ip, port, quorum
101 ) -> Awaitable[bool]: ...
103 def sentinel_monitor(self, name, ip, port, quorum) -> bool | Awaitable[bool]:
104 """Add a new master to Sentinel to be monitored"""
105 return self.execute_command("SENTINEL MONITOR", name, ip, port, quorum)
107 @overload
108 def sentinel_remove(self: SyncClientProtocol, name) -> bool: ...
110 @overload
111 def sentinel_remove(self: AsyncClientProtocol, name) -> Awaitable[bool]: ...
113 def sentinel_remove(self, name) -> bool | Awaitable[bool]:
114 """Remove a master from Sentinel's monitoring"""
115 return self.execute_command("SENTINEL REMOVE", name)
117 @overload
118 def sentinel_sentinels(
119 self: SyncClientProtocol,
120 service_name,
121 return_responses: bool = False,
122 ) -> list[dict[str, Any]] | bool: ...
124 @overload
125 def sentinel_sentinels(
126 self: AsyncClientProtocol,
127 service_name,
128 return_responses: bool = False,
129 ) -> Awaitable[list[dict[str, Any]] | bool]: ...
131 def sentinel_sentinels(self, service_name, return_responses: bool = False) -> (
132 list[dict[str, Any]] | bool
133 ) | Awaitable[list[dict[str, Any]] | bool]:
134 """
135 Returns a list of sentinels for ``service_name``, when return_responses is True,
136 otherwise returns a boolean value that indicates if the command was successful.
137 """
138 return self.execute_command(
139 "SENTINEL SENTINELS", service_name, return_responses=return_responses
140 )
142 @overload
143 def sentinel_set(self: SyncClientProtocol, name, option, value) -> bool: ...
145 @overload
146 def sentinel_set(
147 self: AsyncClientProtocol, name, option, value
148 ) -> Awaitable[bool]: ...
150 def sentinel_set(self, name, option, value) -> bool | Awaitable[bool]:
151 """Set Sentinel monitoring parameters for a given master"""
152 return self.execute_command("SENTINEL SET", name, option, value)
154 @overload
155 def sentinel_slaves(
156 self: SyncClientProtocol, service_name
157 ) -> list[dict[str, Any]]: ...
159 @overload
160 def sentinel_slaves(
161 self: AsyncClientProtocol, service_name
162 ) -> Awaitable[list[dict[str, Any]]]: ...
164 def sentinel_slaves(
165 self,
166 service_name,
167 ) -> list[dict[str, Any]] | Awaitable[list[dict[str, Any]]]:
168 """
169 Returns a list of slaves for ``service_name``
171 Important: This function is called by the Sentinel implementation and is
172 called directly on the Redis standalone client for sentinels,
173 so it doesn't support the "once" and "return_responses" options.
174 """
175 return self.execute_command("SENTINEL SLAVES", service_name)
177 @overload
178 def sentinel_reset(self: SyncClientProtocol, pattern) -> bool: ...
180 @overload
181 def sentinel_reset(self: AsyncClientProtocol, pattern) -> Awaitable[bool]: ...
183 def sentinel_reset(self, pattern) -> bool | Awaitable[bool]:
184 """
185 This command will reset all the masters with matching name.
186 The pattern argument is a glob-style pattern.
188 The reset process clears any previous state in a master (including a
189 failover in progress), and removes every slave and sentinel already
190 discovered and associated with the master.
191 """
192 return self.execute_command("SENTINEL RESET", pattern, once=True)
194 @overload
195 def sentinel_failover(self: SyncClientProtocol, new_master_name) -> bool: ...
197 @overload
198 def sentinel_failover(
199 self: AsyncClientProtocol, new_master_name
200 ) -> Awaitable[bool]: ...
202 def sentinel_failover(self, new_master_name) -> bool | Awaitable[bool]:
203 """
204 Force a failover as if the master was not reachable, and without
205 asking for agreement to other Sentinels (however a new version of the
206 configuration will be published so that the other Sentinels will
207 update their configurations).
208 """
209 return self.execute_command("SENTINEL FAILOVER", new_master_name)
211 @overload
212 def sentinel_ckquorum(self: SyncClientProtocol, new_master_name) -> bool: ...
214 @overload
215 def sentinel_ckquorum(
216 self: AsyncClientProtocol, new_master_name
217 ) -> Awaitable[bool]: ...
219 def sentinel_ckquorum(self, new_master_name) -> bool | Awaitable[bool]:
220 """
221 Check if the current Sentinel configuration is able to reach the
222 quorum needed to failover a master, and the majority needed to
223 authorize the failover.
225 This command should be used in monitoring systems to check if a
226 Sentinel deployment is ok.
227 """
228 return self.execute_command("SENTINEL CKQUORUM", new_master_name, once=True)
230 @overload
231 def sentinel_flushconfig(self: SyncClientProtocol) -> bool: ...
233 @overload
234 def sentinel_flushconfig(self: AsyncClientProtocol) -> Awaitable[bool]: ...
236 def sentinel_flushconfig(self) -> bool | Awaitable[bool]:
237 """
238 Force Sentinel to rewrite its configuration on disk, including the
239 current Sentinel state.
241 Normally Sentinel rewrites the configuration every time something
242 changes in its state (in the context of the subset of the state which
243 is persisted on disk across restart).
244 However sometimes it is possible that the configuration file is lost
245 because of operation errors, disk failures, package upgrade scripts or
246 configuration managers. In those cases a way to to force Sentinel to
247 rewrite the configuration file is handy.
249 This command works even if the previous configuration file is
250 completely missing.
251 """
252 return self.execute_command("SENTINEL FLUSHCONFIG")
255class AsyncSentinelCommands(SentinelCommands):
256 async def sentinel(self, *args) -> None:
257 """Redis Sentinel's SENTINEL command."""
258 super().sentinel(*args)