refactor: common JSON document loader

Author: jg-rp

jsonpath/_data.py

@@ -0,0 +1,21 @@
+import json
+import re
+from io import IOBase
+from typing import Any
+
+_RE_PROBABLY_MALFORMED = re.compile(r"[\{\}\[\]]")
+
+
+def load_data(data: object) -> Any:
+    if isinstance(data, str):
+        try:
+            return json.loads(data)
+        except json.JSONDecodeError:
+            # Overly simple way to detect a malformed JSON document vs a
+            # top-level string only document
+            if _RE_PROBABLY_MALFORMED.search(data):
+                raise
+            return data
+    if isinstance(data, IOBase):
+        return json.loads(data.read())
+    return data

jsonpath/patch.py

@@ -6,25 +6,24 @@
 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
 
-from .exceptions import JSONPatchError
-from .exceptions import JSONPatchTestFailure
-from .exceptions import JSONPointerError
-from .exceptions import JSONPointerIndexError
-from .exceptions import JSONPointerKeyError
-from .exceptions import JSONPointerTypeError
-from .pointer import UNDEFINED
-from .pointer import JSONPointer
+from jsonpath._data import load_data
+from jsonpath.exceptions import JSONPatchError
+from jsonpath.exceptions import JSONPatchTestFailure
+from jsonpath.exceptions import JSONPointerError
+from jsonpath.exceptions import JSONPointerIndexError
+from jsonpath.exceptions import JSONPointerKeyError
+from jsonpath.exceptions import JSONPointerTypeError
+from jsonpath.pointer import UNDEFINED
+from jsonpath.pointer import JSONPointer
 
 
 class Op(ABC):
@@ -537,7 +536,7 @@ def apply(
             JSONPatchTestFailure: When a _test_ operation does not pass.
                 `JSONPatchTestFailure` is a subclass of `JSONPatchError`.
         """
-        _data = _load_data(data)
+        _data = load_data(data)
 
         for i, op in enumerate(self.ops):
             try:
@@ -598,18 +597,3 @@ 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

@@ -2,8 +2,6 @@
 from __future__ import annotations
 
 import itertools
-import json
-from io import IOBase
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import AsyncIterable
@@ -16,10 +14,13 @@
 from typing import TypeVar
 from typing import Union
 
-from .match import FilterContextVars
-from .match import JSONPathMatch
+from jsonpath._data import load_data
+from jsonpath.match import FilterContextVars
+from jsonpath.match import JSONPathMatch
 
 if TYPE_CHECKING:
+    from io import IOBase
+
     from .env import JSONPathEnvironment
     from .selectors import JSONPathSelector
 
@@ -114,7 +115,7 @@ def finditer(
             JSONPathTypeError: If a filter expression attempts to use types in
                 an incompatible way.
         """
-        _data = _load_data(data)
+        _data = load_data(data)
         matches: Iterable[JSONPathMatch] = [
             JSONPathMatch(
                 filter_context=filter_context or {},
@@ -152,7 +153,7 @@ async def finditer_async(
         filter_context: Optional[FilterContextVars] = None,
     ) -> AsyncIterable[JSONPathMatch]:
         """An async version of `finditer()`."""
-        _data = _load_data(data)
+        _data = load_data(data)
 
         async def root_iter() -> AsyncIterable[JSONPathMatch]:
             yield self.env.match_class(
@@ -408,16 +409,3 @@ 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

@@ -2,10 +2,8 @@
 from __future__ import annotations
 
 import codecs
-import json
 import re
 from functools import reduce
-from io import IOBase
 from operator import getitem
 from typing import TYPE_CHECKING
 from typing import Any
@@ -16,6 +14,7 @@
 from typing import Union
 from urllib.parse import unquote
 
+from jsonpath._data import load_data
 from jsonpath.exceptions import JSONPointerError
 from jsonpath.exceptions import JSONPointerIndexError
 from jsonpath.exceptions import JSONPointerKeyError
@@ -25,6 +24,8 @@
 from jsonpath.exceptions import RelativeJSONPointerSyntaxError
 
 if TYPE_CHECKING:
+    from io import IOBase
+
     from .match import JSONPathMatch
 
 
@@ -190,7 +191,7 @@ def resolve(
             JSONPointerTypeError: When attempting to resolve a non-index string
                 path part against a sequence, unless a default is given.
         """
-        data = _load_data(data)
+        data = load_data(data)
         try:
             return reduce(self._getitem, self.parts, data)
         except JSONPointerResolutionError:
@@ -224,7 +225,7 @@ def resolve_parent(
         if not self.parts:
             return (None, self.resolve(data))
 
-        _data = _load_data(data)
+        _data = load_data(data)
         parent = reduce(self._getitem, self.parts[:-1], _data)
 
         try:
@@ -594,19 +595,6 @@ 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]],