-
-
Notifications
You must be signed in to change notification settings - Fork 487
refactor: introduce JsonPathStr type for JSONPath validation #2264
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 6 commits
f9ff86e
c3aa85c
0ccfa7f
1ca5db4
20a89d8
6b0d684
6b12803
d73c012
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,6 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from typing import Any, override | ||
| from typing import Annotated, Any, override | ||
|
|
||
| from giskard.core import NOT_PROVIDED, NotProvided | ||
| from jsonpath_ng import ( | ||
|
|
@@ -15,11 +15,36 @@ | |
| WhereNot, | ||
| parse, | ||
| ) | ||
| from pydantic import BaseModel, Field | ||
| from jsonpath_ng.exceptions import JsonPathLexerError, JsonPathParserError | ||
| from pydantic import AfterValidator, BaseModel, Field | ||
|
|
||
| from .trace import Trace | ||
|
|
||
|
|
||
| class _JsonPathStrMarker: | ||
| """Marker placed in JsonPathStr metadata. Used by the enforcement test.""" | ||
|
|
||
|
|
||
| _REQUIRED_JSONPATH_PREFIX = "trace." | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm wondering: what's the use of forcing each key to start with
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. to make clear that is coming from the trace |
||
|
|
||
|
|
||
| def _validate_jsonpath_syntax(v: str) -> str: | ||
| if not v.startswith(_REQUIRED_JSONPATH_PREFIX): | ||
| raise ValueError( | ||
| f"Invalid JSONPath expression {v!r}: path must start with 'trace.'" | ||
| ) | ||
| try: | ||
| parse(v) | ||
| return v | ||
| except (JsonPathLexerError, JsonPathParserError) as e: | ||
| raise ValueError(f"Invalid JSONPath expression {v!r}: {e}") from e | ||
|
|
||
|
|
||
| JsonPathStr = Annotated[ | ||
| str, AfterValidator(_validate_jsonpath_syntax), _JsonPathStrMarker() | ||
| ] | ||
|
|
||
|
|
||
| class NoMatch(BaseModel): | ||
| """Indicates that a key was provided but no match was found during extraction. | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| """Unit tests for JSONPath validation in extraction.py.""" | ||
|
|
||
| import pytest | ||
| from giskard.checks.core.extraction import JsonPathStr, _validate_jsonpath_syntax | ||
| from pydantic import BaseModel, ValidationError | ||
|
|
||
|
|
||
| class TestValidateJsonpathSyntax: | ||
| """Tests for the _validate_jsonpath_syntax validator function.""" | ||
|
|
||
| def test_valid_simple_path(self): | ||
| assert _validate_jsonpath_syntax("trace.last.outputs") == "trace.last.outputs" | ||
|
|
||
| def test_valid_index_path(self): | ||
| assert ( | ||
| _validate_jsonpath_syntax("trace.interactions[-1].outputs") | ||
| == "trace.interactions[-1].outputs" | ||
| ) | ||
|
|
||
| def test_valid_wildcard_path(self): | ||
| assert ( | ||
| _validate_jsonpath_syntax("trace.interactions[*].outputs") | ||
| == "trace.interactions[*].outputs" | ||
| ) | ||
|
|
||
| def test_valid_nested_path(self): | ||
| assert ( | ||
| _validate_jsonpath_syntax("trace.last.metadata.context") | ||
| == "trace.last.metadata.context" | ||
| ) | ||
|
|
||
| def test_valid_metadata_reference(self): | ||
| assert ( | ||
| _validate_jsonpath_syntax("trace.last.metadata.reference_text") | ||
| == "trace.last.metadata.reference_text" | ||
| ) | ||
|
|
||
| def test_syntax_error_unclosed_bracket(self): | ||
| with pytest.raises(ValueError, match="Invalid JSONPath expression"): | ||
| _validate_jsonpath_syntax("trace.last.outputs[") | ||
|
|
||
| def test_syntax_error_double_bracket(self): | ||
| with pytest.raises(ValueError, match="Invalid JSONPath expression"): | ||
| _validate_jsonpath_syntax("trace.last.outputs[[") | ||
|
|
||
| def test_missing_trace_prefix(self): | ||
| with pytest.raises(ValueError, match="path must start with 'trace\\.'"): | ||
| _validate_jsonpath_syntax("last.outputs") | ||
|
|
||
| def test_root_typo(self): | ||
| with pytest.raises(ValueError, match="path must start with 'trace\\.'"): | ||
| _validate_jsonpath_syntax("tras.last.outputs") | ||
|
|
||
| def test_empty_string_missing_prefix(self): | ||
| with pytest.raises(ValueError, match="path must start with 'trace\\.'"): | ||
| _validate_jsonpath_syntax("") | ||
|
|
||
| def test_just_prefix_is_invalid(self): | ||
| """JSONPath 'trace.' alone should fail parsing as it's incomplete.""" | ||
| with pytest.raises(ValueError, match="Invalid JSONPath expression"): | ||
| _validate_jsonpath_syntax("trace.") | ||
|
|
||
| def test_prefix_is_case_sensitive(self): | ||
| """Prefix must be lowercase 'trace.', not 'Trace.' or 'TRACE.'""" | ||
| with pytest.raises(ValueError, match="path must start with 'trace\\.'"): | ||
| _validate_jsonpath_syntax("Trace.last.outputs") | ||
|
|
||
| def test_whitespace_prefix_is_invalid(self): | ||
| """Leading whitespace should be rejected.""" | ||
| with pytest.raises(ValueError, match="path must start with 'trace\\.'"): | ||
| _validate_jsonpath_syntax(" trace.last.outputs") | ||
|
|
||
| def test_valid_descendants_operator(self): | ||
| """JSONPath descendants operator (..) should be valid.""" | ||
| assert _validate_jsonpath_syntax("trace..outputs") == "trace..outputs" | ||
|
|
||
|
|
||
| class TestJsonPathStrAnnotatedType: | ||
| """Tests for JsonPathStr as a Pydantic Annotated field type.""" | ||
|
|
||
| def test_valid_path_accepted(self): | ||
| class Model(BaseModel): | ||
| key: JsonPathStr | ||
|
|
||
| m = Model(key="trace.last.outputs") | ||
| assert m.key == "trace.last.outputs" | ||
|
|
||
| def test_invalid_syntax_raises_validation_error(self): | ||
| class Model(BaseModel): | ||
| key: JsonPathStr | ||
|
|
||
| with pytest.raises(ValidationError, match="Invalid JSONPath expression"): | ||
| Model(key="trace.last.outputs[") | ||
|
|
||
| def test_missing_trace_prefix_raises_validation_error(self): | ||
| class Model(BaseModel): | ||
| key: JsonPathStr | ||
|
|
||
| with pytest.raises(ValidationError, match="path must start with 'trace\\.'"): | ||
| Model(key="last.outputs") | ||
|
|
||
| def test_optional_jsonpath_str_accepts_none(self): | ||
| class Model(BaseModel): | ||
| key: JsonPathStr | None = None | ||
|
|
||
| m = Model(key=None) | ||
| assert m.key is None | ||
|
|
||
| def test_optional_jsonpath_str_validates_when_provided(self): | ||
| """Optional fields should still validate when a string value is provided.""" | ||
|
|
||
| class Model(BaseModel): | ||
| key: JsonPathStr | None | ||
|
|
||
| with pytest.raises(ValidationError, match="Invalid JSONPath expression"): | ||
| Model(key="trace.last.outputs[") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| """Enforcement test: all JSONPath fields in Check subclasses must use JsonPathStr.""" | ||
|
|
||
| import re | ||
| import types | ||
| from typing import Annotated, Union, get_args, get_origin | ||
|
|
||
| import giskard.checks.builtin # noqa: F401 - triggers all @Check.register imports | ||
| from giskard.checks.core.check import Check | ||
| from giskard.checks.core.extraction import _JsonPathStrMarker | ||
|
|
||
| JSONPATH_FIELD = re.compile(r"^(key|.+_key)$") | ||
|
|
||
|
|
||
| def _all_check_subclasses(cls): | ||
| """Recursively yield all concrete and abstract subclasses of cls. | ||
|
|
||
| This ensures all Check subclasses follow the JsonPathStr convention, | ||
| even if they're abstract base classes. | ||
| """ | ||
| for sub in cls.__subclasses__(): | ||
| yield sub | ||
| yield from _all_check_subclasses(sub) | ||
|
|
||
|
|
||
| def _annotation_has_marker(annotation) -> bool: | ||
| """Recursively check if an annotation contains _JsonPathStrMarker. | ||
|
|
||
| This handles complex type annotations like: | ||
| - Annotated[str, AfterValidator(...), _JsonPathStrMarker()] | ||
| - JsonPathStr | None | ||
| - JsonPathStr | NotProvided | ||
| """ | ||
| if get_origin(annotation) is Annotated: | ||
| return any(isinstance(m, _JsonPathStrMarker) for m in get_args(annotation)[1:]) | ||
| origin = get_origin(annotation) | ||
| if origin is Union or isinstance(annotation, types.UnionType): | ||
| return any(_annotation_has_marker(arg) for arg in get_args(annotation)) | ||
| return False | ||
|
|
||
|
|
||
| def _has_jsonpath_marker(field_info) -> bool: | ||
| """Return True if the field uses JsonPathStr. | ||
|
|
||
| Pydantic v2 stores Annotated metadata in two places depending on the type: | ||
| - Simple `JsonPathStr`: marker is in field_info.metadata | ||
| - Union `JsonPathStr | None` / `JsonPathStr | NotProvided`: marker is | ||
| inside field_info.annotation (the Annotated[str, ...] is preserved within | ||
| the Union at the annotation level) | ||
| """ | ||
| if any(isinstance(m, _JsonPathStrMarker) for m in field_info.metadata): | ||
| return True | ||
| return _annotation_has_marker(field_info.annotation) | ||
|
|
||
|
|
||
| def test_all_jsonpath_fields_use_jsonpath_str(): | ||
| """Enforce that all JSONPath fields use the JsonPathStr type. | ||
|
|
||
| This architectural constraint ensures: | ||
| 1. JSONPath syntax is validated at model creation time | ||
| 2. All JSONPath expressions start with 'trace.' prefix | ||
| 3. API consistency across all Check subclasses | ||
| 4. Better error messages for users when they provide invalid paths | ||
| """ | ||
| violations = [] | ||
| for cls in _all_check_subclasses(Check): | ||
| if not hasattr(cls, "model_fields"): | ||
| continue | ||
| for field_name, field_info in cls.model_fields.items(): | ||
| if JSONPATH_FIELD.match(field_name): | ||
| if not _has_jsonpath_marker(field_info): | ||
| violations.append( | ||
| f"{cls.__name__}.{field_name}: {field_info.annotation}" | ||
| ) | ||
| assert not violations, ( | ||
| "The following JSONPath fields do not use JsonPathStr.\n" | ||
| "All fields named 'key' or ending in '_key' must be annotated as JsonPathStr " | ||
| "(or JsonPathStr | None / JsonPathStr | NotProvided):\n" | ||
| + "\n".join(f" - {v}" for v in violations) | ||
| ) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Minor thing but for consistency,
JSONPathStrwould be the recommended PEP8 style: