feat: add serialization helpers for SearchIndex and AsyncSearchIndex

Add to_dict() and to_yaml() methods to both SearchIndex and AsyncSearchIndex
classes for improved automation and reproducibility.

Features:
- to_dict() serializes schema configuration to dictionary
- to_yaml() writes configuration to YAML file
- Optional include_connection parameter with security-safe defaults
- Sensitive fields (passwords) are never serialized
- Only safe connection keys are included (ssl, socket_timeout, etc.)

Closes redis#493

Author: thakoreh

redisvl/index/index.py

@@ -1294,6 +1294,49 @@ def info(self, name: Optional[str] = None) -> Dict[str, Any]:
         index_name = name or self.schema.index.name
         return self._info(index_name, self._redis_client)
 
+    def to_dict(self, include_connection: bool = False) -> Dict[str, Any]:
+        """Serialize the index configuration to a dictionary.
+
+        Args:
+            include_connection (bool, optional): Whether to include connection
+                parameters. Defaults to False for security (passwords/URLs
+                are excluded by default).
+
+        Returns:
+            Dict[str, Any]: Dictionary representation of the index configuration.
+
+        Example:
+            >>> config = index.to_dict()
+            >>> new_index = SearchIndex.from_dict(config)
+        """
+        config = self.schema.to_dict()
+        if include_connection:
+            config["_redis_url"] = self._redis_url
+            # Note: connection_kwargs may contain sensitive info
+            # Only include non-sensitive keys
+            safe_keys = {"decode_responses", "ssl", "socket_timeout", "socket_connect_timeout"}
+            config["_connection_kwargs"] = {
+                k: v for k, v in self._connection_kwargs.items()
+                if k in safe_keys
+            }
+        return config
+
+    def to_yaml(self, path: str, include_connection: bool = False) -> None:
+        """Serialize the index configuration to a YAML file.
+
+        Args:
+            path (str): Path to write the YAML file.
+            include_connection (bool, optional): Whether to include connection
+                parameters. Defaults to False for security.
+
+        Example:
+            >>> index.to_yaml("schemas/my_index.yaml")
+        """
+        import yaml
+        config = self.to_dict(include_connection=include_connection)
+        with open(path, "w") as f:
+            yaml.dump(config, f, default_flow_style=False, sort_keys=False)
+
     def __enter__(self):
         return self
 
@@ -2251,6 +2294,49 @@ def disconnect_sync(self):
             return
         sync_wrapper(self.disconnect)()
 
+    def to_dict(self, include_connection: bool = False) -> Dict[str, Any]:
+        """Serialize the index configuration to a dictionary.
+
+        Args:
+            include_connection (bool, optional): Whether to include connection
+                parameters. Defaults to False for security (passwords/URLs
+                are excluded by default).
+
+        Returns:
+            Dict[str, Any]: Dictionary representation of the index configuration.
+
+        Example:
+            >>> config = index.to_dict()
+            >>> new_index = AsyncSearchIndex.from_dict(config)
+        """
+        config = self.schema.to_dict()
+        if include_connection:
+            config["_redis_url"] = self._redis_url
+            # Note: connection_kwargs may contain sensitive info
+            # Only include non-sensitive keys
+            safe_keys = {"decode_responses", "ssl", "socket_timeout", "socket_connect_timeout"}
+            config["_connection_kwargs"] = {
+                k: v for k, v in self._connection_kwargs.items()
+                if k in safe_keys
+            }
+        return config
+
+    def to_yaml(self, path: str, include_connection: bool = False) -> None:
+        """Serialize the index configuration to a YAML file.
+
+        Args:
+            path (str): Path to write the YAML file.
+            include_connection (bool, optional): Whether to include connection
+                parameters. Defaults to False for security.
+
+        Example:
+            >>> await index.to_yaml("schemas/my_index.yaml")
+        """
+        import yaml
+        config = self.to_dict(include_connection=include_connection)
+        with open(path, "w") as f:
+            yaml.dump(config, f, default_flow_style=False, sort_keys=False)
+
     async def __aenter__(self):
         return self
 

tests/unit/test_index_serialization.py

@@ -0,0 +1,166 @@
+"""Tests for SearchIndex serialization helpers."""
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from redisvl.index import AsyncSearchIndex, SearchIndex
+from redisvl.schema import IndexSchema
+
+
+@pytest.fixture
+def sample_schema():
+    """Create a sample schema for testing."""
+    return IndexSchema.from_dict({
+        "index": {
+            "name": "test_index",
+            "prefix": "test:",
+            "storage_type": "hash",
+        },
+        "fields": [
+            {"name": "text", "type": "text"},
+            {"name": "vector", "type": "vector", "attrs": {"dims": 128, "algorithm": "flat"}},
+        ]
+    })
+
+
+class TestSearchIndexSerialization:
+    """Tests for SearchIndex serialization methods."""
+
+    def test_to_dict_without_connection(self, sample_schema):
+        """Test to_dict() excludes connection info by default."""
+        index = SearchIndex(
+            schema=sample_schema,
+            redis_url="redis://localhost:6379",
+        )
+        
+        config = index.to_dict()
+        
+        assert "index" in config
+        assert config["index"]["name"] == "test_index"
+        assert "_redis_url" not in config
+        assert "_connection_kwargs" not in config
+
+    def test_to_dict_with_connection(self, sample_schema):
+        """Test to_dict() includes connection info when requested."""
+        index = SearchIndex(
+            schema=sample_schema,
+            redis_url="redis://localhost:6379",
+            connection_kwargs={"ssl": True, "socket_timeout": 30, "password": "secret"},
+        )
+        
+        config = index.to_dict(include_connection=True)
+        
+        assert "_redis_url" in config
+        assert config["_redis_url"] == "redis://localhost:6379"
+        # Password should not be included (not in safe_keys)
+        assert "password" not in config.get("_connection_kwargs", {})
+        # Safe keys should be included
+        assert config["_connection_kwargs"]["ssl"] is True
+        assert config["_connection_kwargs"]["socket_timeout"] == 30
+
+    def test_to_yaml_without_connection(self, sample_schema):
+        """Test to_yaml() writes schema to YAML file."""
+        index = SearchIndex(schema=sample_schema)
+        
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            path = f.name
+        
+        try:
+            index.to_yaml(path)
+            
+            content = Path(path).read_text()
+            assert "test_index" in content
+            assert "text" in content
+            assert "vector" in content
+        finally:
+            Path(path).unlink()
+
+    def test_to_yaml_with_connection(self, sample_schema):
+        """Test to_yaml() includes connection when requested."""
+        index = SearchIndex(
+            schema=sample_schema,
+            redis_url="redis://localhost:6379",
+        )
+        
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            path = f.name
+        
+        try:
+            index.to_yaml(path, include_connection=True)
+            
+            content = Path(path).read_text()
+            assert "_redis_url" in content
+        finally:
+            Path(path).unlink()
+
+    def test_roundtrip_dict(self, sample_schema):
+        """Test that to_dict() and from_dict() roundtrip correctly."""
+        original_index = SearchIndex(
+            schema=sample_schema,
+            connection_kwargs={"ssl": True},
+        )
+        
+        config = original_index.to_dict()
+        restored_index = SearchIndex.from_dict(config)
+        
+        assert restored_index.schema.index.name == original_index.schema.index.name
+        assert restored_index.schema.index.prefix == original_index.schema.index.prefix
+
+
+class TestAsyncSearchIndexSerialization:
+    """Tests for AsyncSearchIndex serialization methods."""
+
+    def test_to_dict_without_connection(self, sample_schema):
+        """Test to_dict() excludes connection info by default."""
+        index = AsyncSearchIndex(
+            schema=sample_schema,
+            redis_url="redis://localhost:6379",
+        )
+        
+        config = index.to_dict()
+        
+        assert "index" in config
+        assert config["index"]["name"] == "test_index"
+        assert "_redis_url" not in config
+
+    def test_to_dict_with_connection(self, sample_schema):
+        """Test to_dict() includes connection info when requested."""
+        index = AsyncSearchIndex(
+            schema=sample_schema,
+            redis_url="redis://localhost:6379",
+            connection_kwargs={"ssl": True, "password": "secret"},
+        )
+        
+        config = index.to_dict(include_connection=True)
+        
+        assert "_redis_url" in config
+        # Password should not be included (not in safe_keys)
+        assert "password" not in config.get("_connection_kwargs", {})
+
+    def test_to_yaml(self, sample_schema):
+        """Test to_yaml() writes schema to YAML file."""
+        index = AsyncSearchIndex(schema=sample_schema)
+        
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            path = f.name
+        
+        try:
+            index.to_yaml(path)
+            
+            content = Path(path).read_text()
+            assert "test_index" in content
+        finally:
+            Path(path).unlink()
+
+    def test_roundtrip_dict(self, sample_schema):
+        """Test that to_dict() and from_dict() roundtrip correctly."""
+        original_index = AsyncSearchIndex(
+            schema=sample_schema,
+            connection_kwargs={"ssl": True},
+        )
+        
+        config = original_index.to_dict()
+        restored_index = AsyncSearchIndex.from_dict(config)
+        
+        assert restored_index.schema.index.name == original_index.schema.index.name