From 972b78953674ef59ba637f44f3f0fb4c39bcca54 Mon Sep 17 00:00:00 2001 From: yixi ning Date: Sun, 18 Oct 2026 08:18:55 +0000 Subject: [PATCH 1/2] Fix loading data and keys from single-pass iterables --- redisvl/index/storage.py | 12 ++- tests/integration/test_load_iterables.py | 97 ++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 4 deletions(-) create mode 100644 tests/integration/test_load_iterables.py diff --git a/redisvl/index/storage.py b/redisvl/index/storage.py index da9290b4a..43ecdc1b0 100644 --- a/redisvl/index/storage.py +++ b/redisvl/index/storage.py @@ -375,7 +375,9 @@ def write( ValueError: If the length of provided keys does not match the length of objects, or if validation fails. """ - if keys and len(keys) != len(objects): # type: ignore + objects = list(objects) + keys = list(keys) if keys is not None else None + if keys is not None and len(keys) != len(objects): raise ValueError("Length of keys does not match the length of objects") if batch_size is None: @@ -386,7 +388,7 @@ def write( # Pass 1: Preprocess and validate all objects prepared_objects = self._preprocess_and_validate_objects( - list(objects), # Convert Iterable to List + objects, id_field=id_field, keys=keys, preprocess=preprocess, @@ -454,7 +456,9 @@ async def awrite( ValueError: If the length of provided keys does not match the length of objects, or if validation fails. """ - if keys and len(keys) != len(objects): # type: ignore + objects = list(objects) + keys = list(keys) if keys is not None else None + if keys is not None and len(keys) != len(objects): raise ValueError("Length of keys does not match the length of objects") if batch_size is None: @@ -465,7 +469,7 @@ async def awrite( # Pass 1: Preprocess and validate all objects prepared_objects = self._preprocess_and_validate_objects( - list(objects), # Convert Iterable to List + objects, id_field=id_field, keys=keys, preprocess=preprocess, diff --git a/tests/integration/test_load_iterables.py b/tests/integration/test_load_iterables.py new file mode 100644 index 000000000..433cb70ef --- /dev/null +++ b/tests/integration/test_load_iterables.py @@ -0,0 +1,97 @@ +"""Explicit load keys and data may be single-pass iterables.""" + +import pytest + +from redisvl.exceptions import RedisVLError +from redisvl.index import AsyncSearchIndex, SearchIndex +from redisvl.schema import IndexSchema + + +@pytest.fixture(params=["hash", "json"]) +def load_schema(request, redis_test_name): + return IndexSchema.from_dict( + { + "index": { + "name": redis_test_name("iterable_load"), + "prefix": redis_test_name("iterable_load"), + "storage_type": request.param, + }, + "fields": [{"name": "title", "type": "text"}], + } + ) + + +@pytest.mark.parametrize("data_iterator", [False, True]) +@pytest.mark.parametrize("keys_iterator", [False, True]) +def test_load_with_iterable_keys(client, load_schema, data_iterator, keys_iterator): + index = SearchIndex(load_schema, redis_client=client) + records = [{"title": "first"}, {"title": "second"}] + keys = [index.key(str(i)) for i in range(len(records))] + try: + loaded = index.load( + iter(records) if data_iterator else records, + keys=iter(keys) if keys_iterator else keys, + batch_size=1, + ) + assert loaded == keys + assert [index.fetch(str(i)) for i in range(len(records))] == records + finally: + client.delete(*keys) + + +@pytest.mark.parametrize("data_iterator", [False, True]) +@pytest.mark.parametrize("keys_iterator", [False, True]) +async def test_async_load_with_iterable_keys( + async_client, load_schema, data_iterator, keys_iterator +): + index = AsyncSearchIndex(load_schema, redis_client=async_client) + records = [{"title": "first"}, {"title": "second"}] + keys = [index.key(str(i)) for i in range(len(records))] + try: + loaded = await index.load( + iter(records) if data_iterator else records, + keys=iter(keys) if keys_iterator else keys, + batch_size=1, + ) + assert loaded == keys + assert [await index.fetch(str(i)) for i in range(len(records))] == records + finally: + await async_client.delete(*keys) + + +@pytest.mark.parametrize("key_count", [0, 1, 3]) +def test_load_rejects_mismatched_keys_before_writing(client, load_schema, key_count): + index = SearchIndex(load_schema, redis_client=client) + keys = [index.key(str(i)) for i in range(key_count)] + try: + with pytest.raises(RedisVLError, match="Length of keys does not match"): + index.load( + iter([{"title": "first"}, {"title": "second"}]), + keys=iter(keys) if keys else [], + batch_size=1, + ) + assert list(client.scan_iter(f"{index.prefix}:*")) == [] + finally: + stored_keys = list(client.scan_iter(f"{index.prefix}:*")) + if stored_keys: + client.delete(*stored_keys) + + +@pytest.mark.parametrize("key_count", [0, 1, 3]) +async def test_async_load_rejects_mismatched_keys_before_writing( + async_client, load_schema, key_count +): + index = AsyncSearchIndex(load_schema, redis_client=async_client) + keys = [index.key(str(i)) for i in range(key_count)] + try: + with pytest.raises(RedisVLError, match="Length of keys does not match"): + await index.load( + iter([{"title": "first"}, {"title": "second"}]), + keys=iter(keys) if keys else [], + batch_size=1, + ) + assert [key async for key in async_client.scan_iter(f"{index.prefix}:*")] == [] + finally: + stored_keys = [key async for key in async_client.scan_iter(f"{index.prefix}:*")] + if stored_keys: + await async_client.delete(*stored_keys) From 92b6298db6007451acbb3598d24bf3b4d8c39c3f Mon Sep 17 00:00:00 2001 From: yixi ning Date: Sat, 14 Nov 2026 02:07:40 +0800 Subject: [PATCH 2/2] test(index): replace load iterables integration test with hermetic unit tests Per review, cover single-pass iterable support and mismatched-keys validation in BaseStorage.write/awrite with small Redis-free tests in tests/unit/test_storage.py. --- tests/integration/test_load_iterables.py | 97 ------------------------ tests/unit/test_storage.py | 30 +++++++- 2 files changed, 29 insertions(+), 98 deletions(-) delete mode 100644 tests/integration/test_load_iterables.py diff --git a/tests/integration/test_load_iterables.py b/tests/integration/test_load_iterables.py deleted file mode 100644 index 433cb70ef..000000000 --- a/tests/integration/test_load_iterables.py +++ /dev/null @@ -1,97 +0,0 @@ -"""Explicit load keys and data may be single-pass iterables.""" - -import pytest - -from redisvl.exceptions import RedisVLError -from redisvl.index import AsyncSearchIndex, SearchIndex -from redisvl.schema import IndexSchema - - -@pytest.fixture(params=["hash", "json"]) -def load_schema(request, redis_test_name): - return IndexSchema.from_dict( - { - "index": { - "name": redis_test_name("iterable_load"), - "prefix": redis_test_name("iterable_load"), - "storage_type": request.param, - }, - "fields": [{"name": "title", "type": "text"}], - } - ) - - -@pytest.mark.parametrize("data_iterator", [False, True]) -@pytest.mark.parametrize("keys_iterator", [False, True]) -def test_load_with_iterable_keys(client, load_schema, data_iterator, keys_iterator): - index = SearchIndex(load_schema, redis_client=client) - records = [{"title": "first"}, {"title": "second"}] - keys = [index.key(str(i)) for i in range(len(records))] - try: - loaded = index.load( - iter(records) if data_iterator else records, - keys=iter(keys) if keys_iterator else keys, - batch_size=1, - ) - assert loaded == keys - assert [index.fetch(str(i)) for i in range(len(records))] == records - finally: - client.delete(*keys) - - -@pytest.mark.parametrize("data_iterator", [False, True]) -@pytest.mark.parametrize("keys_iterator", [False, True]) -async def test_async_load_with_iterable_keys( - async_client, load_schema, data_iterator, keys_iterator -): - index = AsyncSearchIndex(load_schema, redis_client=async_client) - records = [{"title": "first"}, {"title": "second"}] - keys = [index.key(str(i)) for i in range(len(records))] - try: - loaded = await index.load( - iter(records) if data_iterator else records, - keys=iter(keys) if keys_iterator else keys, - batch_size=1, - ) - assert loaded == keys - assert [await index.fetch(str(i)) for i in range(len(records))] == records - finally: - await async_client.delete(*keys) - - -@pytest.mark.parametrize("key_count", [0, 1, 3]) -def test_load_rejects_mismatched_keys_before_writing(client, load_schema, key_count): - index = SearchIndex(load_schema, redis_client=client) - keys = [index.key(str(i)) for i in range(key_count)] - try: - with pytest.raises(RedisVLError, match="Length of keys does not match"): - index.load( - iter([{"title": "first"}, {"title": "second"}]), - keys=iter(keys) if keys else [], - batch_size=1, - ) - assert list(client.scan_iter(f"{index.prefix}:*")) == [] - finally: - stored_keys = list(client.scan_iter(f"{index.prefix}:*")) - if stored_keys: - client.delete(*stored_keys) - - -@pytest.mark.parametrize("key_count", [0, 1, 3]) -async def test_async_load_rejects_mismatched_keys_before_writing( - async_client, load_schema, key_count -): - index = AsyncSearchIndex(load_schema, redis_client=async_client) - keys = [index.key(str(i)) for i in range(key_count)] - try: - with pytest.raises(RedisVLError, match="Length of keys does not match"): - await index.load( - iter([{"title": "first"}, {"title": "second"}]), - keys=iter(keys) if keys else [], - batch_size=1, - ) - assert [key async for key in async_client.scan_iter(f"{index.prefix}:*")] == [] - finally: - stored_keys = [key async for key in async_client.scan_iter(f"{index.prefix}:*")] - if stored_keys: - await async_client.delete(*stored_keys) diff --git a/tests/unit/test_storage.py b/tests/unit/test_storage.py index e79c9650d..c5fbe4ff8 100644 --- a/tests/unit/test_storage.py +++ b/tests/unit/test_storage.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest from pydantic import ValidationError @@ -181,3 +181,31 @@ def fn(d): preprocessed_data = storage_instance._preprocess(obj=data, preprocess=fn) assert "foo" in preprocessed_data assert preprocessed_data["foo"] == "bar" + + +ITERABLE_SCHEMA = IndexSchema.from_dict( + { + "index": {"name": "t", "prefix": "t"}, + "fields": [{"name": "title", "type": "text"}], + } +) +ITERABLE_OBJS = [{"title": "a"}, {"title": "b"}] + + +async def test_write_accepts_single_pass_iterables(): + storage = HashStorage(index_schema=ITERABLE_SCHEMA) + keys = ["t:1", "t:2"] + assert storage.write(MagicMock(), iter(ITERABLE_OBJS), keys=iter(keys)) == keys + client = MagicMock() + client.pipeline.return_value.__aenter__.return_value = AsyncMock() + assert await storage.awrite(client, iter(ITERABLE_OBJS), keys=iter(keys)) == keys + + +@pytest.mark.parametrize("keys", [[], ["t:1"], ["t:1", "t:2", "t:3"]]) +async def test_write_rejects_mismatched_keys_before_writing(keys): + storage, client = HashStorage(index_schema=ITERABLE_SCHEMA), MagicMock() + with pytest.raises(ValueError, match="Length of keys"): + storage.write(client, iter(ITERABLE_OBJS), keys=iter(keys)) + with pytest.raises(ValueError, match="Length of keys"): + await storage.awrite(client, iter(ITERABLE_OBJS), keys=iter(keys)) + client.pipeline.assert_not_called()