refactor(validation): DRY field checks, clearer root path, null test

Author: clean6378-max-it

tests/test_jsonl_validation.py

@@ -39,6 +39,12 @@ def test_wrong_type_session_id(self):
             validate_session_dict(_valid_payload(session_id=123))
         assert exc_info.value.path == "session_id"
 
+    def test_null_session_id(self):
+        with pytest.raises(SessionValidationError) as exc_info:
+            validate_session_dict(_valid_payload(session_id=None))
+        assert exc_info.value.path == "session_id"
+        assert exc_info.value.detail == "must not be null"
+
     def test_null_role_in_message(self):
         with pytest.raises(SessionValidationError) as exc_info:
             validate_session_dict(

utils/validation.py

@@ -5,68 +5,52 @@
 from models.errors import SessionValidationError
 from models.session import SessionDict
 
-_REQUIRED_SESSION_KEYS = ("session_id", "title", "messages", "metadata")
+_ROOT_PATH = "(root)"
+
+
+def _require_field(
+    container: dict[str, Any],
+    key: str,
+    expected_type: type[Any],
+    type_label: str,
+    *,
+    path: str | None = None,
+) -> Any:
+    field_path = path or key
+    if key not in container:
+        raise SessionValidationError(field_path, "missing required field")
+    return _require_value(field_path, container[key], expected_type, type_label)
+
+
+def _require_value(
+    path: str,
+    val: Any,
+    expected_type: type[Any],
+    type_label: str,
+) -> Any:
+    if val is None:
+        raise SessionValidationError(path, "must not be null")
+    if not isinstance(val, expected_type):
+        raise SessionValidationError(
+            path, f"expected {type_label}, got {type(val).__name__}"
+        )
+    return val
 
 
 def validate_session_dict(data: dict[str, Any]) -> SessionDict:
     """Validate a plain dict matches SessionDict before returning it."""
     # Runtime guard for dynamic callers; mypy already types the parameter as dict.
     if not isinstance(data, dict):
-        raise SessionValidationError("$", "expected dict")
-
-    for key in _REQUIRED_SESSION_KEYS:
-        if key not in data:
-            raise SessionValidationError(key, "missing required field")
-
-    session_id = data["session_id"]
-    # Explicit null check before isinstance so errors say "must not be null".
-    if session_id is None:
-        raise SessionValidationError("session_id", "must not be null")
-    if not isinstance(session_id, str):
-        raise SessionValidationError(
-            "session_id", f"expected str, got {type(session_id).__name__}"
-        )
-
-    title = data["title"]
-    if title is None:
-        raise SessionValidationError("title", "must not be null")
-    if not isinstance(title, str):
-        raise SessionValidationError(
-            "title", f"expected str, got {type(title).__name__}"
-        )
+        raise SessionValidationError(_ROOT_PATH, "expected dict")
 
-    messages = data["messages"]
-    if messages is None:
-        raise SessionValidationError("messages", "must not be null")
-    if not isinstance(messages, list):
-        raise SessionValidationError(
-            "messages", f"expected list, got {type(messages).__name__}"
-        )
+    _require_field(data, "session_id", str, "str")
+    _require_field(data, "title", str, "str")
+    messages = _require_field(data, "messages", list, "list")
+    _require_field(data, "metadata", dict, "dict")
 
     for index, message in enumerate(messages):
         path = f"messages[{index}]"
-        if message is None:
-            raise SessionValidationError(path, "must not be null")
-        if not isinstance(message, dict):
-            raise SessionValidationError(
-                path, f"expected dict, got {type(message).__name__}"
-            )
-        if "role" not in message:
-            raise SessionValidationError(f"{path}.role", "missing required field")
-        role = message["role"]
-        if role is None:
-            raise SessionValidationError(f"{path}.role", "must not be null")
-        if not isinstance(role, str):
-            raise SessionValidationError(
-                f"{path}.role", f"expected str, got {type(role).__name__}"
-            )
-
-    metadata = data["metadata"]
-    if metadata is None:
-        raise SessionValidationError("metadata", "must not be null")
-    if not isinstance(metadata, dict):
-        raise SessionValidationError(
-            "metadata", f"expected dict, got {type(metadata).__name__}"
-        )
+        msg_dict = _require_value(path, message, dict, "dict")
+        _require_field(msg_dict, "role", str, "str", path=f"{path}.role")
 
     return cast(SessionDict, data)