fix: handling of top-level string only JSON documents

Author: jg-rp

CHANGELOG.md

@@ -9,6 +9,7 @@
 **Fixes**
 
 - Fixed a bug with the parsing of JSON Pointers. When given an arbitrary string without slashes, `JSONPointer` would resolve to the document root. The empty string is the only valid pointer that should resolve to the document root. We now raise a `JSONPointerError` in such cases. See [#27](https://github.com/jg-rp/python-jsonpath/issues/27).
+- Fixed handling of JSON documents containing only a top-level string.
 
 **Features**
 

jsonpath/patch.py

@@ -6,12 +6,14 @@
 from abc import ABC
 from abc import abstractmethod
 from io import IOBase
+from typing import Any
 from typing import Dict
 from typing import Iterable
 from typing import List
 from typing import Mapping
 from typing import MutableMapping
 from typing import MutableSequence
+from typing import Sequence
 from typing import TypeVar
 from typing import Union
 
@@ -513,7 +515,7 @@ def test(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
     def apply(
         self,
         data: Union[str, IOBase, MutableSequence[object], MutableMapping[str, object]],
-    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
+    ) -> object:
         """Apply all operations from this patch to _data_.
 
         If _data_ is a string or file-like object, it will be loaded with
@@ -535,14 +537,7 @@ def apply(
             JSONPatchTestFailure: When a _test_ operation does not pass.
                 `JSONPatchTestFailure` is a subclass of `JSONPatchError`.
         """
-        if isinstance(data, str):
-            _data: Union[
-                MutableSequence[object], MutableMapping[str, object]
-            ] = json.loads(data)
-        elif isinstance(data, IOBase):
-            _data = json.loads(data.read())
-        else:
-            _data = data
+        _data = _load_data(data)
 
         for i, op in enumerate(self.ops):
             try:
@@ -603,3 +598,18 @@ def apply(
         unicode_escape=unicode_escape,
         uri_decode=uri_decode,
     ).apply(data)
+
+
+def _load_data(
+    data: Union[int, str, IOBase, Sequence[Any], MutableMapping[str, Any]]
+) -> Any:
+    if isinstance(data, str):
+        try:
+            return json.loads(data)
+        except json.JSONDecodeError:
+            data = data.strip()
+            if data.startswith('"') and data.endswith('"'):
+                return data
+    if isinstance(data, IOBase):
+        return json.loads(data.read())
+    return data

jsonpath/path.py

@@ -114,15 +114,7 @@ def finditer(
             JSONPathTypeError: If a filter expression attempts to use types in
                 an incompatible way.
         """
-        # TODO: _load_data()
-        #  - possibly a scalar value
-        if isinstance(data, str):
-            _data = json.loads(data)
-        elif isinstance(data, IOBase):
-            _data = json.loads(data.read())
-        else:
-            _data = data
-
+        _data = _load_data(data)
         matches: Iterable[JSONPathMatch] = [
             JSONPathMatch(
                 filter_context=filter_context or {},
@@ -160,12 +152,7 @@ async def finditer_async(
         filter_context: Optional[FilterContextVars] = None,
     ) -> AsyncIterable[JSONPathMatch]:
         """An async version of `finditer()`."""
-        if isinstance(data, str):
-            _data = json.loads(data)
-        elif isinstance(data, IOBase):
-            _data = json.loads(data.read())
-        else:
-            _data = data
+        _data = _load_data(data)
 
         async def root_iter() -> AsyncIterable[JSONPathMatch]:
             yield self.env.match_class(
@@ -421,3 +408,16 @@ async def _achain(*iterables: AsyncIterable[T]) -> AsyncIterable[T]:
     for it in iterables:
         async for element in it:
             yield element
+
+
+def _load_data(data: Union[str, IOBase, Sequence[Any], Mapping[str, Any]]) -> Any:
+    if isinstance(data, str):
+        try:
+            return json.loads(data)
+        except json.JSONDecodeError:
+            data = data.strip()
+            if data.startswith('"') and data.endswith('"'):
+                return data
+    if isinstance(data, IOBase):
+        return json.loads(data.read())
+    return data

jsonpath/pointer.py

@@ -190,10 +190,7 @@ def resolve(
             JSONPointerTypeError: When attempting to resolve a non-index string
                 path part against a sequence, unless a default is given.
         """
-        if isinstance(data, str):
-            data = json.loads(data)
-        elif isinstance(data, IOBase):
-            data = json.loads(data.read())
+        data = _load_data(data)
         try:
             return reduce(self._getitem, self.parts, data)
         except JSONPointerResolutionError:
@@ -227,13 +224,7 @@ def resolve_parent(
         if not self.parts:
             return (None, self.resolve(data))
 
-        if isinstance(data, str):
-            _data = json.loads(data)
-        elif isinstance(data, IOBase):
-            _data = json.loads(data.read())
-        else:
-            _data = data
-
+        _data = _load_data(data)
         parent = reduce(self._getitem, self.parts[:-1], _data)
 
         try:
@@ -603,6 +594,19 @@ def to(
         )
 
 
+def _load_data(data: Union[int, str, IOBase, Sequence[Any], Mapping[str, Any]]) -> Any:
+    if isinstance(data, str):
+        try:
+            return json.loads(data)
+        except json.JSONDecodeError:
+            data = data.strip()
+            if data.startswith('"') and data.endswith('"'):
+                return data
+    if isinstance(data, IOBase):
+        return json.loads(data.read())
+    return data
+
+
 def resolve(
     pointer: Union[str, Iterable[Union[str, int]]],
     data: Union[str, IOBase, Sequence[object], Mapping[str, object]],

jsonpath/selectors.py

@@ -164,7 +164,7 @@ def resolve(self, matches: Iterable[JSONPathMatch]) -> Iterable[JSONPathMatch]:
                     )
                     match.add_child(_match)
                     yield _match
-            elif isinstance(match.obj, Sequence):
+            elif isinstance(match.obj, Sequence) and not isinstance(match.obj, str):
                 norm_index = self._normalized_index(match.obj)
                 with suppress(IndexError):
                     _match = self.env.match_class(
@@ -195,7 +195,7 @@ async def resolve_async(
                     )
                     match.add_child(_match)
                     yield _match
-            elif isinstance(match.obj, Sequence):
+            elif isinstance(match.obj, Sequence) and not isinstance(match.obj, str):
                 norm_index = self._normalized_index(match.obj)
                 with suppress(IndexError):
                     _match = self.env.match_class(

tests/consensus.py

@@ -45,7 +45,6 @@ class Query:
 
 SKIP = {
     "bracket_notation_with_number_on_object": "Bad consensus",
-    "bracket_notation_with_number_on_string": "Invalid document",
     "dot_notation_with_number_-1": "Unexpected token",
     "dot_notation_with_number_on_object": "conflict with compliance",
 }