Introduce strict mode and use regex if available

Author: jg-rp

jsonpath/env.py

@@ -88,7 +88,8 @@ class attributes `root_token`, `self_token` and `filter_context_token`.
             well-typedness as compile time.
 
             **New in version 0.10.0**
-
+        strict: When `True`, follow RFC 9535 strictly.
+            **New in version 2.0.0**
     ## Class attributes
 
     Attributes:
@@ -143,6 +144,7 @@ def __init__(
         filter_caching: bool = True,
         unicode_escape: bool = True,
         well_typed: bool = True,
+        strict: bool = False,
     ) -> None:
         self.filter_caching: bool = filter_caching
         """Enable or disable filter expression caching."""
@@ -154,6 +156,14 @@ def __init__(
         self.well_typed: bool = well_typed
         """Control well-typedness checks on filter function expressions."""
 
+        self.strict: bool = strict
+        """When `True`, follow RFC 9535 strictly.
+        
+        This includes things like enforcing a leading root identifier and
+        ensuring there's not leading or trailing whitespace when parsing a
+        JSONPath query.
+        """
+
         self.lexer: Lexer = self.lexer_class(env=self)
         """The lexer bound to this environment."""
 
@@ -188,8 +198,10 @@ def compile(self, path: str) -> Union[JSONPath, CompoundJSONPath]:  # noqa: A003
             env=self, segments=self.parser.parse(stream), pseudo_root=pseudo_root
         )
 
-        # TODO: Optionally raise for trailing whitespace
-        stream.skip_whitespace()
+        if stream.skip_whitespace() and self.strict:
+            raise JSONPathSyntaxError(
+                "unexpected whitespace", token=stream.tokens[stream.pos - 1]
+            )
 
         # TODO: better!
         if stream.current().kind != TOKEN_EOF:

jsonpath/function_extensions/match.py

@@ -1,6 +1,9 @@
 """The standard `match` function extension."""
 
-import re
+try:
+    import regex as re
+except ImportError:
+    import re  # type: ignore
 
 from jsonpath.function_extensions import ExpressionType
 from jsonpath.function_extensions import FilterFunction

jsonpath/function_extensions/search.py

@@ -1,6 +1,9 @@
 """The standard `search` function extension."""
 
-import re
+try:
+    import regex as re
+except ImportError:
+    import re  # type: ignore
 
 from jsonpath.function_extensions import ExpressionType
 from jsonpath.function_extensions import FilterFunction

jsonpath/parse.py

@@ -101,6 +101,7 @@
 from .token import TOKEN_TRUE
 from .token import TOKEN_UNDEFINED
 from .token import TOKEN_UNION
+from .token import TOKEN_WHITESPACE
 from .token import TOKEN_WILD
 from .token import Token
 
@@ -295,10 +296,29 @@ def __init__(self, *, env: JSONPathEnvironment) -> None:
         }
 
     def parse(self, stream: TokenStream) -> Iterator[JSONPathSegment]:
-        """Parse a JSONPath from a stream of tokens."""
-        # TODO: Optionally require TOKEN_ROOT
-        if stream.current().kind in {TOKEN_ROOT, TOKEN_PSEUDO_ROOT}:
+        """Parse a JSONPath query from a stream of tokens."""
+        if stream.skip_whitespace() and self.env.strict:
+            raise JSONPathSyntaxError(
+                "unexpected leading whitespace", token=stream.current()
+            )
+
+        if (
+            self.env.strict
+            and len(stream.tokens)
+            and stream.tokens[-1].kind == TOKEN_WHITESPACE
+        ):
+            raise JSONPathSyntaxError(
+                "unexpected trailing whitespace", token=stream.tokens[-1]
+            )
+
+        token = stream.current()
+
+        if token.kind == TOKEN_ROOT or (
+            token.kind == TOKEN_PSEUDO_ROOT and not self.env.strict
+        ):
             stream.next()
+        elif self.env.strict:
+            stream.expect(TOKEN_ROOT)
 
         yield from self.parse_query(stream)
 

jsonpath/selectors.py

@@ -130,15 +130,15 @@ def _normalized_index(self, obj: Sequence[object]) -> int:
         return self.index
 
     def resolve(self, node: JSONPathMatch) -> Iterable[JSONPathMatch]:
-        # TODO: Optionally try string representation of int
-        # if isinstance(node.obj, Mapping):
-        #     # Try the string representation of the index as a key.
-        #     with suppress(KeyError):
-        #         match = node.new_child(
-        #             self.env.getitem(node.obj, self._as_key), self.index
-        #         )
-        #         node.add_child(match)
-        #         yield match
+        # Optionally try string representation of int
+        if not self.env.strict and isinstance(node.obj, Mapping):
+            # Try the string representation of the index as a key.
+            with suppress(KeyError):
+                match = node.new_child(
+                    self.env.getitem(node.obj, self._as_key), self.index
+                )
+                node.add_child(match)
+                yield match
         if isinstance(node.obj, Sequence) and not isinstance(node.obj, str):
             norm_index = self._normalized_index(node.obj)
             with suppress(IndexError):
@@ -149,15 +149,14 @@ def resolve(self, node: JSONPathMatch) -> Iterable[JSONPathMatch]:
                 yield match
 
     async def resolve_async(self, node: JSONPathMatch) -> AsyncIterable[JSONPathMatch]:
-        # XXX
-        # if isinstance(node.obj, Mapping):
-        #     # Try the string representation of the index as a key.
-        #     with suppress(KeyError):
-        #         match = node.new_child(
-        #             await self.env.getitem_async(node.obj, self._as_key), self.index
-        #         )
-        #         node.add_child(match)
-        #         yield match
+        if not self.env.strict and isinstance(node.obj, Mapping):
+            # Try the string representation of the index as a key.
+            with suppress(KeyError):
+                match = node.new_child(
+                    await self.env.getitem_async(node.obj, self._as_key), self.index
+                )
+                node.add_child(match)
+                yield match
         if isinstance(node.obj, Sequence) and not isinstance(node.obj, str):
             norm_index = self._normalized_index(node.obj)
             with suppress(IndexError):

jsonpath/stream.py

@@ -89,7 +89,9 @@ def expect_peek_not(self, typ: str, message: str) -> None:
         if self.peek().kind == typ:
             raise JSONPathSyntaxError(message, token=self.peek())
 
-    def skip_whitespace(self) -> None:
+    def skip_whitespace(self) -> bool:
         """Skip whitespace."""
         if self.current().kind == TOKEN_WHITESPACE:
             self.pos += 1
+            return True
+        return False

pyproject.toml

@@ -48,11 +48,12 @@ include = ["/jsonpath"]
 dependencies = [
   "pytest",
   "pytest-cov",
-  "black",
   "mypy",
-  "ipython",
+  "regex",
+  "iregexp-check",
   "pyyaml",
   "types-pyyaml",
+  "types-regex",
   "twine",
   "ruff",
 ]
@@ -78,6 +79,9 @@ dependencies = ["black", "mkdocs", "mkdocstrings[python]", "mkdocs-material"]
 build = "mkdocs build --clean --strict"
 serve = "mkdocs serve --dev-addr localhost:8000"
 
+[tool.hatch.envs.no-regex]
+dependencies = ["pytest"]
+
 [tool.coverage.run]
 branch = true
 parallel = true

tests/test_compliance.py

@@ -18,7 +18,9 @@
 
 import pytest
 
-import jsonpath
+from jsonpath import JSONPathEnvironment
+from jsonpath import JSONPathError
+from jsonpath import NodeList
 
 
 @dataclass
@@ -35,10 +37,6 @@ class Case:
 
 
 SKIP = {
-    # "basic, no leading whitespace": "flexible whitespace policy",
-    "basic, no trailing whitespace": "flexible whitespace policy",
-    # "basic, bald descendant segment": "almost has a consensus",
-    # "filter, index segment on object, selects nothing": "flexible selector policy",
     "functions, match, dot matcher on \\u2028": "standard library re policy",
     "functions, match, dot matcher on \\u2029": "standard library re policy",
     "functions, search, dot matcher on \\u2028": "standard library re policy",
@@ -76,14 +74,6 @@ class Case:
     "name selector, double quotes, non-surrogate surrogate": "expected behavior policy",
     "name selector, double quotes, surrogate supplementary": "expected behavior policy",
     "name selector, double quotes, supplementary surrogate": "expected behavior policy",
-    # "whitespace, selectors, space between dot and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, newline between dot and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, tab between dot and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, return between dot and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, space between recursive descent and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, newline between recursive descent and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, tab between recursive descent and name": "flexible whitespace policy",  # noqa: E501
-    # "whitespace, selectors, return between recursive descent and name": "flexible whitespace policy",  # noqa: E501
 }
 
 
@@ -101,13 +91,18 @@ def invalid_cases() -> List[Case]:
     return [case for case in cases() if case.invalid_selector]
 
 
+@pytest.fixture()
+def env() -> JSONPathEnvironment:
+    return JSONPathEnvironment(strict=True)
+
+
 @pytest.mark.parametrize("case", valid_cases(), ids=operator.attrgetter("name"))
-def test_compliance(case: Case) -> None:
+def test_compliance(env: JSONPathEnvironment, case: Case) -> None:
     if case.name in SKIP:
         pytest.skip(reason=SKIP[case.name])
 
     assert case.document is not None
-    nodes = jsonpath.NodeList(jsonpath.finditer(case.selector, case.document))
+    nodes = NodeList(env.finditer(case.selector, case.document))
 
     if case.results is not None:
         assert case.results_paths is not None
@@ -120,14 +115,14 @@ def test_compliance(case: Case) -> None:
 
 
 @pytest.mark.parametrize("case", valid_cases(), ids=operator.attrgetter("name"))
-def test_compliance_async(case: Case) -> None:
+def test_compliance_async(env: JSONPathEnvironment, case: Case) -> None:
     if case.name in SKIP:
         pytest.skip(reason=SKIP[case.name])
 
-    async def coro() -> jsonpath.NodeList:
+    async def coro() -> NodeList:
         assert case.document is not None
-        it = await jsonpath.finditer_async(case.selector, case.document)
-        return jsonpath.NodeList([node async for node in it])
+        it = await env.finditer_async(case.selector, case.document)
+        return NodeList([node async for node in it])
 
     nodes = asyncio.run(coro())
 
@@ -142,9 +137,9 @@ async def coro() -> jsonpath.NodeList:
 
 
 @pytest.mark.parametrize("case", invalid_cases(), ids=operator.attrgetter("name"))
-def test_invalid_selectors(case: Case) -> None:
+def test_invalid_selectors(env: JSONPathEnvironment, case: Case) -> None:
     if case.name in SKIP:
         pytest.skip(reason=SKIP[case.name])
 
-    with pytest.raises(jsonpath.JSONPathError):
-        jsonpath.compile(case.selector)
+    with pytest.raises(JSONPathError):
+        env.compile(case.selector)

tests/test_strictness.py

@@ -0,0 +1,26 @@
+import pytest
+
+from jsonpath import JSONPathEnvironment
+
+
+@pytest.fixture()
+def env() -> JSONPathEnvironment:
+    return JSONPathEnvironment(strict=False)
+
+
+def test_leading_whitespace(env: JSONPathEnvironment) -> None:
+    query = "  $.a"
+    data = {"a": 1}
+    assert env.findall(query, data) == [1]
+
+
+def test_trailing_whitespace(env: JSONPathEnvironment) -> None:
+    query = "$.a  "
+    data = {"a": 1}
+    assert env.findall(query, data) == [1]
+
+
+def test_index_as_object_name(env: JSONPathEnvironment) -> None:
+    query = "$.a[0]"
+    data = {"a": {"0": 1}}
+    assert env.findall(query, data) == [1]