Skip to content

AsyncSentinelClient

Bases: object

Asynchronous Redis Sentinel Client.

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

Source code in pyredis/client/async_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
class AsyncSentinelClient(object):
    """
    Asynchronous Redis Sentinel Client.

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

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

        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

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

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

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

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

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

        Args:
            *args: Command name and arguments.

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

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

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

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

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

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

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

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

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

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

sentinels property

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

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

Initialize the AsyncSentinelClient.

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/async_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 AsyncSentinelClient.

    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() async

Close the active Sentinel connection asynchronously.

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

execute(*args) async

Execute a Sentinel command asynchronously, 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/async_sentinel.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
async def execute(self, *args):
    """
    Execute a Sentinel command asynchronously, automatically selecting an active Sentinel node.

    Args:
        *args: Command name and arguments.

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

get_master(name) async

Get the master node configuration for the specified service name asynchronously.

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/async_sentinel.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def get_master(self, name):
    """
    Get the master node configuration for the specified service name asynchronously.

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

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

get_masters() async

Get configurations for all monitored Redis master nodes asynchronously.

Returns:

Type Description

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

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

get_slaves(name) async

Get replication replica configurations for the specified master service name asynchronously.

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/async_sentinel.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def get_slaves(self, name):
    """
    Get replication replica configurations for the specified master service name asynchronously.

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

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

next_sentinel() async

Close the active connection and rotate the Sentinel node list asynchronously.

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