Skip to content

SentinelClient

Bases: object

Synchronous Redis Sentinel Client.

Handles connectivity to Sentinel nodes and master/slave service discovery.

Source code in pyredis/client/sentinel.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class SentinelClient(object):
    """
    Synchronous Redis Sentinel Client.

    Handles connectivity to Sentinel nodes and master/slave service discovery.
    """

    def __init__(self, sentinels, password=None, username=None):
        """
        Initialize the SentinelClient.

        Args:
            sentinels: List of (host, port) tuples representing the Sentinel nodes.
            password: Optional password for Sentinel authentication.
            username: Optional username for Sentinel ACL authentication.
        """
        self._conn = None
        self._sentinels = deque(sentinels)
        self._password = password
        self._username = username

    def _sentinel_connect(self, sentinel):
        host, port = sentinel
        self._conn = pyredis.client.Connection(
            host=host,
            port=port,
            conn_timeout=0.1,
            sentinel=True,
            password=self._password,
            username=self._username,
        )
        try:
            self.execute("PING")
            return True
        except PyRedisConnError:
            self.close()
            return False

    def _sentinel_get(self):
        for sentinel in range(len(self._sentinels)):
            if self._sentinel_connect(self._sentinels[0]):
                return True
            else:
                self._sentinels.rotate(-1)
        raise PyRedisConnError("Could not connect to any sentinel")

    def close(self):
        """Close the active Sentinel connection."""
        if self._conn:
            self._conn.close()
            self._conn = None

    @property
    def sentinels(self):
        """Deque of configured Sentinel (host, port) node addresses."""
        return self._sentinels

    def execute(self, *args):
        """
        Execute a Sentinel command, automatically selecting an active Sentinel node.

        Args:
            *args: Command name and arguments.

        Returns:
            The Sentinel command response.
        """
        if not self._conn:
            self._sentinel_get()
        self._conn.write(*args)
        return self._conn.read()

    def get_master(self, name):
        """
        Get the master node configuration for the specified service name.

        Args:
            name: The service name of the Redis master.

        Returns:
            Dict containing the master's configuration.
        """
        return pyredis.client.dict_from_list(
            self.execute(
                *["SENTINEL", "master", name]
            )
        )

    def get_masters(self):
        """
        Get configurations for all monitored Redis master nodes.

        Returns:
            List of dicts containing configurations for all monitored masters.
        """
        masters = self.execute(
            *["SENTINEL", "masters"]
        )
        result = []
        for master in masters:
            result.append(
                pyredis.client.dict_from_list(master)
            )
        return result

    def get_slaves(self, name):
        """
        Get replication replica configurations for the specified master service name.

        Args:
            name: The service name of the Redis master.

        Returns:
            List of dicts containing replica configurations.
        """
        slaves = self.execute(
            *["SENTINEL", "slaves", name]
        )
        result = []
        for slave in slaves:
            result.append(
                pyredis.client.dict_from_list(slave)
            )
        return result

    def next_sentinel(self):
        """Close the active connection and rotate the Sentinel node list."""
        self.close()
        self._sentinels.rotate(-1)

sentinels property

Deque of configured Sentinel (host, port) node addresses.

__init__(sentinels, password=None, username=None)

Initialize the SentinelClient.

Parameters:

Name Type Description Default
sentinels

List of (host, port) tuples representing the Sentinel nodes.

 required
password

Optional password for Sentinel authentication.

 None
username

Optional username for Sentinel ACL authentication.

 None
Source code in pyredis/client/sentinel.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, sentinels, password=None, username=None):
    """
    Initialize the SentinelClient.

    Args:
        sentinels: List of (host, port) tuples representing the Sentinel nodes.
        password: Optional password for Sentinel authentication.
        username: Optional username for Sentinel ACL authentication.
    """
    self._conn = None
    self._sentinels = deque(sentinels)
    self._password = password
    self._username = username

close()

Close the active Sentinel connection.

Source code in pyredis/client/sentinel.py 
52
53
54
55
56
def close(self):
    """Close the active Sentinel connection."""
    if self._conn:
        self._conn.close()
        self._conn = None

execute(*args)

Execute a Sentinel command, automatically selecting an active Sentinel node.

Parameters:

Name Type Description Default
*args

Command name and arguments.

 ()

Returns:

Type Description

The Sentinel command response.
Source code in pyredis/client/sentinel.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def execute(self, *args):
    """
    Execute a Sentinel command, automatically selecting an active Sentinel node.

    Args:
        *args: Command name and arguments.

    Returns:
        The Sentinel command response.
    """
    if not self._conn:
        self._sentinel_get()
    self._conn.write(*args)
    return self._conn.read()

get_master(name)

Get the master node configuration for the specified service name.

Parameters:

Name Type Description Default
name

The service name of the Redis master.

 required

Returns:

Type Description

Dict containing the master's configuration.
Source code in pyredis/client/sentinel.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_master(self, name):
    """
    Get the master node configuration for the specified service name.

    Args:
        name: The service name of the Redis master.

    Returns:
        Dict containing the master's configuration.
    """
    return pyredis.client.dict_from_list(
        self.execute(
            *["SENTINEL", "master", name]
        )
    )

get_masters()

Get configurations for all monitored Redis master nodes.

Returns:

Type Description

List of dicts containing configurations for all monitored masters.
Source code in pyredis/client/sentinel.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_masters(self):
    """
    Get configurations for all monitored Redis master nodes.

    Returns:
        List of dicts containing configurations for all monitored masters.
    """
    masters = self.execute(
        *["SENTINEL", "masters"]
    )
    result = []
    for master in masters:
        result.append(
            pyredis.client.dict_from_list(master)
        )
    return result

get_slaves(name)

Get replication replica configurations for the specified master service name.

Parameters:

Name Type Description Default
name

The service name of the Redis master.

 required

Returns:

Type Description

List of dicts containing replica configurations.
Source code in pyredis/client/sentinel.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def get_slaves(self, name):
    """
    Get replication replica configurations for the specified master service name.

    Args:
        name: The service name of the Redis master.

    Returns:
        List of dicts containing replica configurations.
    """
    slaves = self.execute(
        *["SENTINEL", "slaves", name]
    )
    result = []
    for slave in slaves:
        result.append(
            pyredis.client.dict_from_list(slave)
        )
    return result

next_sentinel()

Close the active connection and rotate the Sentinel node list.

Source code in pyredis/client/sentinel.py 
131
132
133
134
def next_sentinel(self):
    """Close the active connection and rotate the Sentinel node list."""
    self.close()
    self._sentinels.rotate(-1)