refactor(schema): extract _parse_schema_from_info to module-level function

- Move pure parsing logic from instance method to module-level function
- Update both SchemaRegistry and AsyncSchemaRegistry to use shared function
- Update tests to call module-level function directly
- Eliminates ~30 lines of code duplication

Author: nkanu17

sql_redis/schema.py

@@ -2,10 +2,51 @@
 
 from __future__ import annotations
 
-from typing import Callable
+from typing import TYPE_CHECKING, Callable
 
 import redis
 
+if TYPE_CHECKING:
+    import redis.asyncio as async_redis
+
+
+def _parse_schema_from_info(info: list) -> dict[str, str]:
+    """Parse field types from FT.INFO response.
+
+    This is a pure function with no I/O operations, shared by both
+    sync and async schema registries.
+
+    Args:
+        info: The raw response from FT.INFO command.
+
+    Returns:
+        Dictionary mapping field names to their types (e.g., {"title": "TEXT"}).
+    """
+    schema = {}
+    # Find the 'attributes' section in the info response
+    for i, item in enumerate(info):
+        # Handle bytes or string comparison
+        item_str = item.decode("utf-8") if isinstance(item, bytes) else item
+        if item_str == "attributes":
+            attributes = info[i + 1]
+            for attr in attributes:
+                field_name = None
+                field_type = None
+                # Each attribute is a list like:
+                # [b'identifier', b'title', b'attribute', b'title', b'type', b'TEXT', ...]
+                for j, val in enumerate(attr):
+                    val_str = val.decode("utf-8") if isinstance(val, bytes) else val
+                    if val_str == "attribute" and j + 1 < len(attr):
+                        fn = attr[j + 1]
+                        field_name = fn.decode("utf-8") if isinstance(fn, bytes) else fn
+                    if val_str == "type" and j + 1 < len(attr):
+                        ft = attr[j + 1]
+                        field_type = ft.decode("utf-8") if isinstance(ft, bytes) else ft
+                if field_name and field_type:
+                    schema[field_name] = field_type
+            break
+    return schema
+
 
 class SchemaRegistry:
     """Loads and caches index schemas from Redis.
@@ -33,43 +74,12 @@ def _load_index_schema(self, index_name: str) -> None:
         """Load schema for a single index."""
         try:
             info = self._client.execute_command("FT.INFO", index_name)
-            schema = self._parse_schema_from_info(info)
+            schema = _parse_schema_from_info(info)
             self._schemas[index_name] = schema
         except redis.ResponseError:
             # Index doesn't exist or was deleted
             self._schemas.pop(index_name, None)
 
-    def _parse_schema_from_info(self, info: list) -> dict[str, str]:
-        """Parse field types from FT.INFO response."""
-        schema = {}
-        # Find the 'attributes' section in the info response
-        for i, item in enumerate(info):
-            # Handle bytes or string comparison
-            item_str = item.decode("utf-8") if isinstance(item, bytes) else item
-            if item_str == "attributes":
-                attributes = info[i + 1]
-                for attr in attributes:
-                    field_name = None
-                    field_type = None
-                    # Each attribute is a list like:
-                    # [b'identifier', b'title', b'attribute', b'title', b'type', b'TEXT', ...]
-                    for j, val in enumerate(attr):
-                        val_str = val.decode("utf-8") if isinstance(val, bytes) else val
-                        if val_str == "attribute" and j + 1 < len(attr):
-                            fn = attr[j + 1]
-                            field_name = (
-                                fn.decode("utf-8") if isinstance(fn, bytes) else fn
-                            )
-                        if val_str == "type" and j + 1 < len(attr):
-                            ft = attr[j + 1]
-                            field_type = (
-                                ft.decode("utf-8") if isinstance(ft, bytes) else ft
-                            )
-                    if field_name and field_type:
-                        schema[field_name] = field_type
-                break
-        return schema
-
     def get_field_type(self, index: str, field: str) -> str | None:
         """Get field type for a given index and field.
 
@@ -140,3 +150,66 @@ def process_pending_events(self) -> None:
             self._schemas.pop(idx, None)
             if self._on_change:
                 self._on_change("dropped", idx)
+
+
+class AsyncSchemaRegistry:
+    """Async version of SchemaRegistry for use with redis.asyncio clients.
+
+    Loads and caches index schemas from Redis asynchronously.
+    """
+
+    def __init__(self, redis_client: "async_redis.Redis") -> None:
+        """Initialize with an async Redis client.
+
+        Args:
+            redis_client: An async Redis client (redis.asyncio.Redis).
+        """
+        self._client = redis_client
+        self._schemas: dict[str, dict[str, str]] = {}
+
+    async def load_all(self) -> None:
+        """Load schemas for all indexes on the server.
+
+        Uses asyncio.gather() to load all index schemas concurrently.
+        """
+        import asyncio
+
+        self._schemas.clear()
+        indexes = await self._client.execute_command("FT._LIST")
+        # Decode bytes to strings
+        decoded_indexes = [
+            idx.decode("utf-8") if isinstance(idx, bytes) else idx for idx in indexes
+        ]
+        # Load all schemas concurrently
+        await asyncio.gather(
+            *[self._load_index_schema(name) for name in decoded_indexes]
+        )
+
+    async def _load_index_schema(self, index_name: str) -> None:
+        """Load schema for a single index."""
+        try:
+            info = await self._client.execute_command("FT.INFO", index_name)
+            schema = _parse_schema_from_info(info)
+            self._schemas[index_name] = schema
+        except redis.ResponseError:
+            # Index doesn't exist or was deleted
+            self._schemas.pop(index_name, None)
+
+    def get_field_type(self, index: str, field: str) -> str | None:
+        """Get field type for a given index and field.
+
+        Returns None if index or field is unknown.
+        """
+        schema = self._schemas.get(index, {})
+        return schema.get(field)
+
+    def get_schema(self, index: str) -> dict[str, str]:
+        """Get full schema for an index.
+
+        Returns empty dict if index is unknown.
+        """
+        return self._schemas.get(index, {})
+
+    async def refresh(self, index_name: str) -> None:
+        """Refresh schema for a single index."""
+        await self._load_index_schema(index_name)

tests/test_schema_registry.py

@@ -3,7 +3,7 @@
 import pytest
 import redis
 
-from sql_redis.schema import SchemaRegistry
+from sql_redis.schema import SchemaRegistry, _parse_schema_from_info
 
 
 def _create_test_indexes(redis_client: redis.Redis) -> list[str]:
@@ -222,20 +222,16 @@ def test_load_all_handles_no_indexes(self, redis_client: redis.Redis):
 class TestSchemaRegistryParsing:
     """Tests for schema parsing edge cases."""
 
-    def test_parse_schema_no_attributes_section(self, redis_client: redis.Redis):
+    def test_parse_schema_no_attributes_section(self):
         """_parse_schema_from_info handles response without attributes."""
-        registry = SchemaRegistry(redis_client)
-
         # FT.INFO response without 'attributes' key
         fake_info = ["index_name", "test", "other_key", "value"]
-        schema = registry._parse_schema_from_info(fake_info)
+        schema = _parse_schema_from_info(fake_info)
 
         assert schema == {}
 
-    def test_parse_schema_incomplete_attribute(self, redis_client: redis.Redis):
+    def test_parse_schema_incomplete_attribute(self):
         """_parse_schema_from_info handles attribute without type."""
-        registry = SchemaRegistry(redis_client)
-
         # FT.INFO response with attribute but missing type
         fake_info = [
             "attributes",
@@ -244,7 +240,7 @@ def test_parse_schema_incomplete_attribute(self, redis_client: redis.Redis):
                 ["identifier", "field2", "attribute", "field2", "type", "TEXT"],
             ],
         ]
-        schema = registry._parse_schema_from_info(fake_info)
+        schema = _parse_schema_from_info(fake_info)
 
         # Only field2 should be captured (field1 has no type)
         assert schema == {"field2": "TEXT"}