dynamic pydantic model validation on load

Author: tylerhutcherson

redisvl/exceptions.py

@@ -1,10 +1,32 @@
-class RedisVLException(Exception):
-    """Base RedisVL exception"""
+"""
+RedisVL Exception Classes
 
+This module defines all custom exceptions used throughout the RedisVL library.
+"""
 
-class RedisModuleVersionError(RedisVLException):
-    """Invalid module versions installed"""
 
+class RedisVLError(Exception):
+    """Base exception for all RedisVL errors."""
 
-class RedisSearchError(RedisVLException):
-    """Error while performing a search or aggregate request"""
+    pass
+
+
+class RedisModuleVersionError(RedisVLError):
+    """Error raised when required Redis modules are missing or have incompatible versions."""
+
+    pass
+
+
+class RedisSearchError(RedisVLError):
+    """Error raised for Redis Search specific operations."""
+
+    pass
+
+
+class SchemaValidationError(RedisVLError):
+    """Error when validating data against a schema."""
+
+    def __init__(self, message, index=None):
+        if index is not None:
+            message = f"Validation failed for object at index {index}: {message}"
+        super().__init__(message)

redisvl/index/index.py

@@ -28,7 +28,12 @@
 import redis.asyncio as aredis
 from redis.commands.search.indexDefinition import IndexDefinition
 
-from redisvl.exceptions import RedisModuleVersionError, RedisSearchError
+from redisvl.exceptions import (
+    RedisModuleVersionError,
+    RedisSearchError,
+    RedisVLError,
+    SchemaValidationError,
+)
 from redisvl.index.storage import BaseStorage, HashStorage, JsonStorage
 from redisvl.query import BaseQuery, CountQuery, FilterQuery
 from redisvl.query.filter import FilterExpression
@@ -574,27 +579,8 @@ def load(
             List[str]: List of keys loaded to Redis.
 
         Raises:
-            ValueError: If the length of provided keys does not match the length
-                of objects or if validation fails when validate_on_load is enabled.
-
-        .. code-block:: python
-
-            data = [{"test": "foo"}, {"test": "bar"}]
-
-            # simple case
-            keys = index.load(data)
-
-            # set 360 second ttl policy on data
-            keys = index.load(data, ttl=360)
-
-            # load data with predefined keys
-            keys = index.load(data, keys=["rvl:foo", "rvl:bar"])
-
-            # load data with preprocessing step
-            def add_field(d):
-                d["new_field"] = 123
-                return d
-            keys = index.load(data, preprocess=add_field)
+            SchemaValidationError: If validation fails when validate_on_load is enabled.
+            RedisVLError: If there's an error loading data to Redis.
         """
         try:
             return self._storage.write(
@@ -607,9 +593,14 @@ def add_field(d):
                 batch_size=batch_size,
                 validate=self._validate_on_load,
             )
-        except:
-            logger.exception("Error while loading data to Redis")
+        except SchemaValidationError:
+            # Pass through validation errors directly
+            logger.exception("Schema validation error while loading data")
             raise
+        except Exception as e:
+            # Wrap other errors as general RedisVL errors
+            logger.exception("Error while loading data to Redis")
+            raise RedisVLError(f"Failed to load data: {str(e)}") from e
 
     def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Fetch an object from Redis by id.
@@ -1134,8 +1125,8 @@ async def load(
             List[str]: List of keys loaded to Redis.
 
         Raises:
-            ValueError: If the length of provided keys does not match the
-                length of objects or if validation fails when validate_on_load is enabled.
+            SchemaValidationError: If validation fails when validate_on_load is enabled.
+            RedisVLError: If there's an error loading data to Redis.
 
         .. code-block:: python
 
@@ -1169,9 +1160,14 @@ def add_field(d):
                 batch_size=batch_size,
                 validate=self._validate_on_load,
             )
-        except:
-            logger.exception("Error while loading data to Redis")
+        except SchemaValidationError:
+            # Pass through validation errors directly
+            logger.exception("Schema validation error while loading data")
             raise
+        except Exception as e:
+            # Wrap other errors as general RedisVL errors
+            logger.exception("Error while loading data to Redis")
+            raise RedisVLError(f"Failed to load data: {str(e)}") from e
 
     async def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Asynchronously etch an object from Redis by id. The id is typically

redisvl/index/storage.py

@@ -5,6 +5,7 @@
 from redis.asyncio import Redis as AsyncRedis
 from redis.commands.search.indexDefinition import IndexType
 
+from redisvl.exceptions import SchemaValidationError
 from redisvl.redis.utils import convert_bytes
 from redisvl.schema import IndexSchema
 from redisvl.schema.validation import validate_object
@@ -180,7 +181,8 @@ def _preprocess_and_validate_objects(
             List of tuples (key, processed_obj) for valid objects
 
         Raises:
-            ValueError: If any validation fails with object context
+            SchemaValidationError: If validation fails, with context about which object failed
+            ValueError: If any other processing errors occur
         """
         prepared_objects = []
         keys_iterator = iter(keys) if keys else None
@@ -197,26 +199,22 @@ def _preprocess_and_validate_objects(
                 # Preprocess
                 processed_obj = self._preprocess(obj, preprocess)
 
-                # Basic type validation
-                if not isinstance(processed_obj, dict):
-                    raise ValueError(
-                        f"Object must be a dictionary, got {type(processed_obj).__name__}"
-                    )
-
                 # Schema validation if enabled
                 if validate:
                     processed_obj = self.validate(processed_obj)
 
                 # Store valid object with its key for writing
                 prepared_objects.append((key, processed_obj))
 
+            except ValidationError as e:
+                # Convert Pydantic ValidationError to SchemaValidationError with index context
+                raise SchemaValidationError(str(e), index=i) from e
             except Exception as e:
-                # Enhance error message with object context
+                # Capture other exceptions with context
                 object_id = f"at index {i}"
-                if id_field and isinstance(obj, dict) and id_field in obj:
-                    object_id = f"with {id_field}={obj[id_field]}"
-
-                raise ValueError(f"Validation failed for object {object_id}: {str(e)}")
+                raise ValueError(
+                    f"Error processing object {object_id}: {str(e)}"
+                ) from e
 
         return prepared_objects