Added CLIENT LIST iterator alternative

redis/_parsers/helpers.py

@@ -919,6 +919,8 @@ def _parse_client_info_fields(value):
 
 
 def parse_client_list(response, **options):
+    if options.pop("as_iter", False):
+        return _client_list_iter(response)
     clients = []
     for c in str_if_bytes(response).splitlines():
         client_dict = _parse_client_info_fields(c)
@@ -927,6 +929,22 @@ def parse_client_list(response, **options):
     return clients
 
 
+def _client_list_iter(response):
+    """
+    Yield one CLIENT LIST record at a time instead of building the full
+    ``list[dict]`` up front, so a caller processing records one at a time
+    doesn't need all of them held in memory simultaneously. The full reply
+    is still buffered off the socket before this runs - that part is
+    unavoidable given how RESP bulk-string framing (and hiredis) works -
+    so this only trims the per-record dict/string overhead, not the raw
+    reply size.
+    """
+    for c in str_if_bytes(response).splitlines():
+        client_dict = _parse_client_info_fields(c)
+        if client_dict:
+            yield client_dict
+
+
 def parse_config_get(response, **options):
     response = [str_if_bytes(i) if i is not None else None for i in response]
     return response and pairs_to_dict(response) or {}

redis/commands/core.py

@@ -863,6 +863,57 @@ def client_list(
             args += client_id
         return self.execute_command("CLIENT LIST", *args, **kwargs)
 
+    @overload
+    def client_list_iter(
+        self: SyncClientProtocol,
+        _type: str | None = None,
+        client_id: List[EncodableT] = [],
+        **kwargs,
+    ) -> Iterator[dict[str, str]]: ...
+
+    @overload
+    def client_list_iter(
+        self: AsyncClientProtocol,
+        _type: str | None = None,
+        client_id: List[EncodableT] = [],
+        **kwargs,
+    ) -> Awaitable[Iterator[dict[str, str]]]: ...
+
+    def client_list_iter(
+        self, _type: str | None = None, client_id: List[EncodableT] = [], **kwargs
+    ) -> Iterator[dict[str, str]] | Awaitable[Iterator[dict[str, str]]]:
+        """
+        Like ``client_list()``, but returns an iterator that parses and
+        yields one client record at a time instead of building the full
+        list upfront. This bounds the memory held for parsed records to
+        one record at a time rather than all of them at once, which
+        matters when there are many thousands of connected clients.
+
+        The full CLIENT LIST reply is still read off the socket in one
+        piece before this can start yielding - that part is unavoidable
+        given how RESP bulk-string framing works - so this only reduces
+        the memory used to hold the parsed records, not the raw reply.
+
+        :param _type: optional. one of the client types (normal, master,
+         replica, pubsub)
+        :param client_id: optional. a list of client ids
+
+        For more information, see https://redis.io/commands/client-list
+        """
+        args = []
+        if _type is not None:
+            client_types = ("normal", "master", "replica", "pubsub")
+            if str(_type).lower() not in client_types:
+                raise DataError(f"CLIENT LIST _type must be one of {client_types!r}")
+            args.append(b"TYPE")
+            args.append(_type)
+        if not isinstance(client_id, list):
+            raise DataError("client_id must be a list")
+        if client_id:
+            args.append(b"ID")
+            args += client_id
+        return self.execute_command("CLIENT LIST", *args, as_iter=True, **kwargs)
+
     @overload
     def client_getname(self: SyncClientProtocol, **kwargs) -> bytes | str | None: ...
 

tests/test_asyncio/test_commands.py

@@ -505,6 +505,12 @@ async def test_client_list(self, r: redis.Redis):
         assert isinstance(clients[0], dict)
         assert "addr" in clients[0]
 
+    @pytest.mark.onlynoncluster
+    async def test_client_list_iter(self, r: redis.Redis):
+        clients = list(await r.client_list_iter())
+        assert isinstance(clients[0], dict)
+        assert "addr" in clients[0]
+
     @skip_if_server_version_lt("5.0.0")
     async def test_client_list_type(self, r: redis.Redis):
         with pytest.raises(exceptions.RedisError):

tests/test_commands.py

@@ -660,6 +660,12 @@ def test_client_list(self, r):
         assert isinstance(clients[0], dict)
         assert "addr" in clients[0]
 
+    @pytest.mark.onlynoncluster
+    def test_client_list_iter(self, r):
+        clients = list(r.client_list_iter())
+        assert isinstance(clients[0], dict)
+        assert "addr" in clients[0]
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("6.2.0")
     def test_client_info(self, r):

tests/test_parsers/test_helpers.py

@@ -1,3 +1,5 @@
+import inspect
+
 import pytest
 
 from redis._parsers.helpers import (
@@ -100,6 +102,16 @@ def test_parse_client_list():
     assert clients == expected
 
 
+@pytest.mark.fixed_client
+def test_parse_client_list_as_iter():
+    # ``as_iter=True`` (client_list_iter's opt-in path) must yield the same
+    # records as the eager list form, just lazily and one at a time.
+    response = "id=1 addr=127.0.0.1:1\nid=2 addr=127.0.0.1:2"
+    result = parse_client_list(response, as_iter=True)
+    assert inspect.isgenerator(result)
+    assert list(result) == parse_client_list(response)
+
+
 @pytest.mark.fixed_client
 def test_parse_client_info():
     # A CLIENT INFO value can contain both a space (a unix-socket addr such as