From 6adf547f174fa4e2235e4ae9e710229b4aee2c9b Mon Sep 17 00:00:00 2001 From: Aryan Pardeshi Date: Sun, 9 Aug 2026 03:08:53 +0530 Subject: [PATCH 1/3] feat: add iter()/aiter() for lazy filter-based key iteration --- redisvl/index/index.py | 74 +++++++++++++++ tests/conftest.py | 31 ++++--- tests/integration/test_index_iteration.py | 106 ++++++++++++++++++++++ uv.lock | 2 +- 4 files changed, 200 insertions(+), 13 deletions(-) create mode 100644 tests/integration/test_index_iteration.py diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 3daa69d3e..4e9949ab7 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -2003,6 +2003,43 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: # Increment the offset for the next batch of pagination offset += page_size + def iter( + self, + filter_expression: str | FilterExpression | None = None, + batch_size: int = DEFAULT_BULK_BATCH_SIZE, + ) -> Generator[str, None, None]: + """Iterate lazily over document keys matching a filter expression. + + Args: + filter_expression (Union[str, FilterExpression, None]): Selects the + documents to iterate over. Defaults to None (all documents). + batch_size (int): Number of keys fetched per query batch. Defaults to 500. + + Yields: + str: Document key matching the filter. + """ + filter_expr = ( + FilterExpression("*") + if filter_expression is None + else filter_expression + ) + query = FilterQuery(filter_expr, return_fields=["id"]) + offset = 0 + while True: + query.paging(offset, batch_size) + batch = self._query(query) + if not batch: + break + for record in batch: + yield record["id"] + offset += len(batch) + if len(batch) < batch_size: + break + + def __iter__(self) -> Generator[str, None, None]: + """Yield document keys in the index.""" + return self.iter() + def listall(self) -> list[str]: """List all search indices in Redis database. @@ -3246,6 +3283,43 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato yield results first += page_size + async def aiter( + self, + filter_expression: str | FilterExpression | None = None, + batch_size: int = DEFAULT_BULK_BATCH_SIZE, + ) -> AsyncGenerator[str, None]: + """Iterate lazily over document keys matching a filter expression asynchronously. + + Args: + filter_expression (Union[str, FilterExpression, None]): Selects the + documents to iterate over. Defaults to None (all documents). + batch_size (int): Number of keys fetched per query batch. Defaults to 500. + + Yields: + str: Document key matching the filter. + """ + filter_expr = ( + FilterExpression("*") + if filter_expression is None + else filter_expression + ) + query = FilterQuery(filter_expr, return_fields=["id"]) + offset = 0 + while True: + query.paging(offset, batch_size) + batch = await self._query(query) + if not batch: + break + for record in batch: + yield record["id"] + offset += len(batch) + if len(batch) < batch_size: + break + + def __aiter__(self) -> AsyncGenerator[str, None]: + """Yield document keys in the index asynchronously.""" + return self.aiter() + async def listall(self) -> list[str]: """List all search indices in Redis database. diff --git a/tests/conftest.py b/tests/conftest.py index cd09ef5d4..302eb5fd4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -64,16 +64,18 @@ def redis_container(worker_id): os.environ["COMPOSE_PROJECT_NAME"] = f"redis_test_{worker_id}" os.environ.setdefault("REDIS_IMAGE", "redis:8.4") - compose = DockerCompose( - context="tests", - compose_file_name="docker-compose.yml", - pull=True, - ) - compose.start() - - yield compose - - compose.stop() + try: + compose = DockerCompose( + context="tests", + compose_file_name="docker-compose.yml", + pull=True, + ) + compose.start() + yield compose + compose.stop() + except Exception as e: + logger.warning(f"DockerCompose failed to start, falling back to local Redis: {e}") + yield None @pytest.fixture(scope="session") @@ -184,8 +186,13 @@ def redis_url(redis_container): Use the `DockerCompose` fixture to get host/port of the 'redis' service on container port 6379 (mapped to an ephemeral port on the host). """ - host, port = redis_container.get_service_host_and_port("redis", 6379) - return f"redis://{host}:{port}" + if redis_container is not None: + try: + host, port = redis_container.get_service_host_and_port("redis", 6379) + return f"redis://{host}:{port}" + except Exception: + pass + return os.getenv("REDIS_URL", "redis://localhost:6379") @pytest.fixture(scope="session") diff --git a/tests/integration/test_index_iteration.py b/tests/integration/test_index_iteration.py new file mode 100644 index 000000000..1d117da23 --- /dev/null +++ b/tests/integration/test_index_iteration.py @@ -0,0 +1,106 @@ +import pytest +from redisvl.index import SearchIndex, AsyncSearchIndex +from redisvl.query.filter import Tag + + +@pytest.fixture +def sample_index(redis_url, redis_test_name): + index_name = redis_test_name("iter_index") + prefix = redis_test_name("iter_doc") + index = SearchIndex.from_dict( + { + "index": {"name": index_name, "prefix": prefix, "storage_type": "hash"}, + "fields": [{"name": "category", "type": "tag"}], + }, + redis_url=redis_url, + ) + index.create(overwrite=True) + docs = [ + {"id": f"{prefix}:1", "category": "A"}, + {"id": f"{prefix}:2", "category": "B"}, + {"id": f"{prefix}:3", "category": "A"}, + {"id": f"{prefix}:4", "category": "C"}, + ] + index.load(docs) + yield index + index.delete(drop=True) + + +@pytest.fixture +async def async_sample_index(redis_url, redis_test_name): + index_name = redis_test_name("async_iter_index") + prefix = redis_test_name("async_iter_doc") + index = AsyncSearchIndex.from_dict( + { + "index": {"name": index_name, "prefix": prefix, "storage_type": "hash"}, + "fields": [{"name": "category", "type": "tag"}], + }, + redis_url=redis_url, + ) + await index.create(overwrite=True) + docs = [ + {"id": f"{prefix}:1", "category": "A"}, + {"id": f"{prefix}:2", "category": "B"}, + {"id": f"{prefix}:3", "category": "A"}, + {"id": f"{prefix}:4", "category": "C"}, + ] + await index.load(docs) + yield index + await index.delete(drop=True) + + +def test_sync_index_iter(sample_index): + # Iterate all keys + all_keys = list(sample_index.iter()) + assert len(all_keys) == 4 + assert set(all_keys) == { + f"{sample_index.prefix}:1", + f"{sample_index.prefix}:2", + f"{sample_index.prefix}:3", + f"{sample_index.prefix}:4", + } + + # Iterate with filter + filter_a = Tag("category") == "A" + filtered_keys = list(sample_index.iter(filter_expression=filter_a)) + assert len(filtered_keys) == 2 + assert set(filtered_keys) == { + f"{sample_index.prefix}:1", + f"{sample_index.prefix}:3", + } + + # Magic __iter__ + magic_keys = list(sample_index) + assert len(magic_keys) == 4 + + +@pytest.mark.asyncio +async def test_async_index_aiter(async_sample_index): + # Iterate all keys asynchronously + all_keys = [] + async for key in async_sample_index.aiter(): + all_keys.append(key) + assert len(all_keys) == 4 + assert set(all_keys) == { + f"{async_sample_index.prefix}:1", + f"{async_sample_index.prefix}:2", + f"{async_sample_index.prefix}:3", + f"{async_sample_index.prefix}:4", + } + + # Iterate with filter asynchronously + filter_a = Tag("category") == "A" + filtered_keys = [] + async for key in async_sample_index.aiter(filter_expression=filter_a): + filtered_keys.append(key) + assert len(filtered_keys) == 2 + assert set(filtered_keys) == { + f"{async_sample_index.prefix}:1", + f"{async_sample_index.prefix}:3", + } + + # Magic __aiter__ + magic_keys = [] + async for key in async_sample_index: + magic_keys.append(key) + assert len(magic_keys) == 4 diff --git a/uv.lock b/uv.lock index 8936cf407..210d06189 100644 --- a/uv.lock +++ b/uv.lock @@ -4841,7 +4841,7 @@ wheels = [ [[package]] name = "redisvl" -version = "0.25.0" +version = "0.25.1" source = { editable = "." } dependencies = [ { name = "jsonpath-ng" }, From 441a51d53f42eb70a06e60b2d2f0014de9d199c1 Mon Sep 17 00:00:00 2001 From: Aryan Pardeshi Date: Sun, 9 Aug 2026 11:39:54 +0530 Subject: [PATCH 2/3] fix: correct iteration tests to use deterministic keys, drop unrequested __iter__/__aiter__, restore conftest --- redisvl/index/index.py | 16 +-- tests/conftest.py | 31 +++--- tests/integration/test_index_iteration.py | 126 +++++++++++----------- uv.lock | 2 +- 4 files changed, 81 insertions(+), 94 deletions(-) diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 4e9949ab7..7ae284f90 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -2019,9 +2019,7 @@ def iter( str: Document key matching the filter. """ filter_expr = ( - FilterExpression("*") - if filter_expression is None - else filter_expression + FilterExpression("*") if filter_expression is None else filter_expression ) query = FilterQuery(filter_expr, return_fields=["id"]) offset = 0 @@ -2036,10 +2034,6 @@ def iter( if len(batch) < batch_size: break - def __iter__(self) -> Generator[str, None, None]: - """Yield document keys in the index.""" - return self.iter() - def listall(self) -> list[str]: """List all search indices in Redis database. @@ -3299,9 +3293,7 @@ async def aiter( str: Document key matching the filter. """ filter_expr = ( - FilterExpression("*") - if filter_expression is None - else filter_expression + FilterExpression("*") if filter_expression is None else filter_expression ) query = FilterQuery(filter_expr, return_fields=["id"]) offset = 0 @@ -3316,10 +3308,6 @@ async def aiter( if len(batch) < batch_size: break - def __aiter__(self) -> AsyncGenerator[str, None]: - """Yield document keys in the index asynchronously.""" - return self.aiter() - async def listall(self) -> list[str]: """List all search indices in Redis database. diff --git a/tests/conftest.py b/tests/conftest.py index 302eb5fd4..cd09ef5d4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -64,18 +64,16 @@ def redis_container(worker_id): os.environ["COMPOSE_PROJECT_NAME"] = f"redis_test_{worker_id}" os.environ.setdefault("REDIS_IMAGE", "redis:8.4") - try: - compose = DockerCompose( - context="tests", - compose_file_name="docker-compose.yml", - pull=True, - ) - compose.start() - yield compose - compose.stop() - except Exception as e: - logger.warning(f"DockerCompose failed to start, falling back to local Redis: {e}") - yield None + compose = DockerCompose( + context="tests", + compose_file_name="docker-compose.yml", + pull=True, + ) + compose.start() + + yield compose + + compose.stop() @pytest.fixture(scope="session") @@ -186,13 +184,8 @@ def redis_url(redis_container): Use the `DockerCompose` fixture to get host/port of the 'redis' service on container port 6379 (mapped to an ephemeral port on the host). """ - if redis_container is not None: - try: - host, port = redis_container.get_service_host_and_port("redis", 6379) - return f"redis://{host}:{port}" - except Exception: - pass - return os.getenv("REDIS_URL", "redis://localhost:6379") + host, port = redis_container.get_service_host_and_port("redis", 6379) + return f"redis://{host}:{port}" @pytest.fixture(scope="session") diff --git a/tests/integration/test_index_iteration.py b/tests/integration/test_index_iteration.py index 1d117da23..968eb867d 100644 --- a/tests/integration/test_index_iteration.py +++ b/tests/integration/test_index_iteration.py @@ -1,7 +1,15 @@ import pytest -from redisvl.index import SearchIndex, AsyncSearchIndex + +from redisvl.index import AsyncSearchIndex, SearchIndex from redisvl.query.filter import Tag +DOCS = [ + {"id": "1", "category": "A"}, + {"id": "2", "category": "B"}, + {"id": "3", "category": "A"}, + {"id": "4", "category": "C"}, +] + @pytest.fixture def sample_index(redis_url, redis_test_name): @@ -15,13 +23,8 @@ def sample_index(redis_url, redis_test_name): redis_url=redis_url, ) index.create(overwrite=True) - docs = [ - {"id": f"{prefix}:1", "category": "A"}, - {"id": f"{prefix}:2", "category": "B"}, - {"id": f"{prefix}:3", "category": "A"}, - {"id": f"{prefix}:4", "category": "C"}, - ] - index.load(docs) + # id_field makes the key deterministic: : + index.load(DOCS, id_field="id") yield index index.delete(drop=True) @@ -38,69 +41,72 @@ async def async_sample_index(redis_url, redis_test_name): redis_url=redis_url, ) await index.create(overwrite=True) - docs = [ - {"id": f"{prefix}:1", "category": "A"}, - {"id": f"{prefix}:2", "category": "B"}, - {"id": f"{prefix}:3", "category": "A"}, - {"id": f"{prefix}:4", "category": "C"}, - ] - await index.load(docs) + await index.load(DOCS, id_field="id") yield index await index.delete(drop=True) -def test_sync_index_iter(sample_index): - # Iterate all keys - all_keys = list(sample_index.iter()) - assert len(all_keys) == 4 - assert set(all_keys) == { - f"{sample_index.prefix}:1", - f"{sample_index.prefix}:2", - f"{sample_index.prefix}:3", - f"{sample_index.prefix}:4", - } +def test_iter_yields_every_key(sample_index): + """iter() with no filter must yield every key in the index, once each.""" + keys = list(sample_index.iter()) + + assert len(keys) == 4 + assert set(keys) == {f"{sample_index.prefix}:{i}" for i in range(1, 5)} + + +def test_iter_respects_filter_expression(sample_index): + """A filter expression must narrow the yielded keys.""" + keys = list(sample_index.iter(filter_expression=Tag("category") == "A")) + + assert set(keys) == {f"{sample_index.prefix}:1", f"{sample_index.prefix}:3"} + + +def test_iter_is_lazy(sample_index): + """Iteration must stream: the first key arrives without draining the index.""" + iterator = sample_index.iter() + + assert next(iterator) is not None - # Iterate with filter - filter_a = Tag("category") == "A" - filtered_keys = list(sample_index.iter(filter_expression=filter_a)) - assert len(filtered_keys) == 2 - assert set(filtered_keys) == { - f"{sample_index.prefix}:1", - f"{sample_index.prefix}:3", - } - # Magic __iter__ - magic_keys = list(sample_index) - assert len(magic_keys) == 4 +def test_iter_pages_when_batch_size_is_smaller_than_the_index(sample_index): + """A batch_size below the document count must still yield every key exactly once.""" + keys = list(sample_index.iter(batch_size=2)) + + assert sorted(keys) == sorted(f"{sample_index.prefix}:{i}" for i in range(1, 5)) @pytest.mark.asyncio -async def test_async_index_aiter(async_sample_index): - # Iterate all keys asynchronously - all_keys = [] - async for key in async_sample_index.aiter(): - all_keys.append(key) - assert len(all_keys) == 4 - assert set(all_keys) == { - f"{async_sample_index.prefix}:1", - f"{async_sample_index.prefix}:2", - f"{async_sample_index.prefix}:3", - f"{async_sample_index.prefix}:4", - } +async def test_aiter_yields_every_key(async_sample_index): + """aiter() must mirror iter() on the async client.""" + keys = [key async for key in async_sample_index.aiter()] - # Iterate with filter asynchronously - filter_a = Tag("category") == "A" - filtered_keys = [] - async for key in async_sample_index.aiter(filter_expression=filter_a): - filtered_keys.append(key) - assert len(filtered_keys) == 2 - assert set(filtered_keys) == { + assert len(keys) == 4 + assert set(keys) == {f"{async_sample_index.prefix}:{i}" for i in range(1, 5)} + + +@pytest.mark.asyncio +async def test_aiter_respects_filter_expression(async_sample_index): + """The async iterator must apply the filter the same way the sync one does.""" + keys = [ + key + async for key in async_sample_index.aiter( + filter_expression=Tag("category") == "A" + ) + ] + + assert set(keys) == { f"{async_sample_index.prefix}:1", f"{async_sample_index.prefix}:3", } - # Magic __aiter__ - magic_keys = [] - async for key in async_sample_index: - magic_keys.append(key) - assert len(magic_keys) == 4 + +@pytest.mark.asyncio +async def test_aiter_pages_when_batch_size_is_smaller_than_the_index( + async_sample_index, +): + """A batch_size below the document count must still yield every key exactly once.""" + keys = [key async for key in async_sample_index.aiter(batch_size=2)] + + assert sorted(keys) == sorted( + f"{async_sample_index.prefix}:{i}" for i in range(1, 5) + ) diff --git a/uv.lock b/uv.lock index 210d06189..8936cf407 100644 --- a/uv.lock +++ b/uv.lock @@ -4841,7 +4841,7 @@ wheels = [ [[package]] name = "redisvl" -version = "0.25.1" +version = "0.25.0" source = { editable = "." } dependencies = [ { name = "jsonpath-ng" }, From 69dd49132ccf2b2eda427d148fd550781f0b128e Mon Sep 17 00:00:00 2001 From: Aryan Pardeshi Date: Mon, 10 Aug 2026 19:57:57 +0530 Subject: [PATCH 3/3] fix: page iter()/aiter() with FT.AGGREGATE cursor instead of FT.SEARCH+LIMIT FT.SEARCH + LIMIT is capped by MAXSEARCHRESULTS and non-deterministic without a unique sort, which is exactly the large-index case this API targets. The repo already has _iter_keys_by_filter for this reason -- it pages with FT.AGGREGATE ... WITHCURSOR and always releases the cursor. Delegate to it instead of reimplementing offset-based paging. Caught by Cursor Bugbot on review, verified against the existing _iter_keys_by_filter docstring and callers (drop_by_filter, update_by_filter). --- redisvl/index/index.py | 45 ++++++++++++++++++------------------------ 1 file changed, 19 insertions(+), 26 deletions(-) diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 7ae284f90..3d0a45e36 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -2010,10 +2010,16 @@ def iter( ) -> Generator[str, None, None]: """Iterate lazily over document keys matching a filter expression. + Delegates to :meth:`_iter_keys_by_filter`, which pages with + ``FT.AGGREGATE ... WITHCURSOR`` rather than ``FT.SEARCH`` + ``LIMIT``, so + this is not subject to the ``MAXSEARCHRESULTS`` limit. See that method's + docstring for why keys are de-duplicated and why memory is + ``O(match count)`` rather than truly streaming. + Args: filter_expression (Union[str, FilterExpression, None]): Selects the documents to iterate over. Defaults to None (all documents). - batch_size (int): Number of keys fetched per query batch. Defaults to 500. + batch_size (int): Number of keys fetched per cursor page. Defaults to 500. Yields: str: Document key matching the filter. @@ -2021,18 +2027,8 @@ def iter( filter_expr = ( FilterExpression("*") if filter_expression is None else filter_expression ) - query = FilterQuery(filter_expr, return_fields=["id"]) - offset = 0 - while True: - query.paging(offset, batch_size) - batch = self._query(query) - if not batch: - break - for record in batch: - yield record["id"] - offset += len(batch) - if len(batch) < batch_size: - break + for batch in self._iter_keys_by_filter(filter_expr, batch_size): + yield from batch def listall(self) -> list[str]: """List all search indices in Redis database. @@ -3284,10 +3280,16 @@ async def aiter( ) -> AsyncGenerator[str, None]: """Iterate lazily over document keys matching a filter expression asynchronously. + Delegates to :meth:`_iter_keys_by_filter`, which pages with + ``FT.AGGREGATE ... WITHCURSOR`` rather than ``FT.SEARCH`` + ``LIMIT``, so + this is not subject to the ``MAXSEARCHRESULTS`` limit. See that method's + docstring for why keys are de-duplicated and why memory is + ``O(match count)`` rather than truly streaming. + Args: filter_expression (Union[str, FilterExpression, None]): Selects the documents to iterate over. Defaults to None (all documents). - batch_size (int): Number of keys fetched per query batch. Defaults to 500. + batch_size (int): Number of keys fetched per cursor page. Defaults to 500. Yields: str: Document key matching the filter. @@ -3295,18 +3297,9 @@ async def aiter( filter_expr = ( FilterExpression("*") if filter_expression is None else filter_expression ) - query = FilterQuery(filter_expr, return_fields=["id"]) - offset = 0 - while True: - query.paging(offset, batch_size) - batch = await self._query(query) - if not batch: - break - for record in batch: - yield record["id"] - offset += len(batch) - if len(batch) < batch_size: - break + async for batch in self._iter_keys_by_filter(filter_expr, batch_size): + for key in batch: + yield key async def listall(self) -> list[str]: """List all search indices in Redis database.