Add Redis scan_keys iterator API.
Test Python Package / unit-tests (pull_request) Successful in 14s
Code Quality Pipeline / code-quality (pull_request) Successful in 32s
Test Python Package / coverage-report (pull_request) Successful in 16s
PR Title Check / check-title (pull_request) Successful in 6s
Test Python Package / integration-tests (pull_request) Successful in 1m3s

Provide a SCAN-based key iterator for Redis adapters so callers can enumerate large keyspaces without relying on blocking KEYS lookups.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Brian Bjarke Jensen
2026-07-08 21:06:13 +02:00
co-authored by Cursor
parent 2f19fcc972
commit 50444af982
6 changed files with 155 additions and 3 deletions
+29
View File
@@ -254,5 +254,34 @@ def test_should_raise_connection_error_on_list_keys_when_not_connected(
adapter.list_keys("some_pattern")
def test_should_scan_keys(
redis_adapter: RedisAdapter,
) -> None:
"""Test scanning keys matching a pattern returns correct keys."""
redis_adapter.set("key1", {"a": 1})
redis_adapter.set("key2", {"b": 2})
keys = set(redis_adapter.scan_keys("key*"))
assert keys == {"key1", "key2"}
def test_should_raise_value_error_on_invalid_scan_keys_pattern(
redis_adapter: RedisAdapter,
) -> None:
"""Test that the RedisAdapter raises ValueError when scanning keys with an invalid pattern."""
invalid_patterns = ["", 123, None]
for pattern in invalid_patterns:
with pytest.raises(ValueError):
list(redis_adapter.scan_keys(pattern)) # type: ignore[arg-type]
def test_should_raise_connection_error_on_scan_keys_when_not_connected(
redis_config: RedisConfig,
) -> None:
"""Test that the RedisAdapter raises ConnectionError when scanning keys while not connected."""
adapter = RedisAdapter(config=redis_config)
with pytest.raises(ConnectionError):
list(adapter.scan_keys("some_pattern"))
if __name__ == "__main__":
pytest.main(["-s", "-v", __file__])