1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365
|
# SPDX-FileCopyrightText: 2023-present James Prior <jamesgr.prior@gmail.com>
#
# SPDX-License-Identifier: MIT
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import AsyncIterable
from typing import Iterable
from typing import List
from typing import Optional
from typing import Union
from ._types import JSON
from ._types import JSONData
from ._types import JSONScalar
from .env import JSONPathEnvironment
from .exceptions import JSONPatchError
from .exceptions import JSONPatchTestFailure
from .exceptions import JSONPathError
from .exceptions import JSONPathIndexError
from .exceptions import JSONPathNameError
from .exceptions import JSONPathSyntaxError
from .exceptions import JSONPathTypeError
from .exceptions import JSONPointerError
from .exceptions import JSONPointerIndexError
from .exceptions import JSONPointerKeyError
from .exceptions import JSONPointerResolutionError
from .exceptions import JSONPointerTypeError
from .exceptions import RelativeJSONPointerError
from .exceptions import RelativeJSONPointerIndexError
from .exceptions import RelativeJSONPointerSyntaxError
from .filter import UNDEFINED
from .fluent_api import Projection
from .fluent_api import Query
from .lex import Lexer
from .match import JSONPathMatch
from .match import NodeList
from .parse import Parser
from .patch import JSONPatch
from .path import CompoundJSONPath
from .path import JSONPath
from .pointer import JSONPointer
from .pointer import RelativeJSONPointer
from .pointer import resolve
if TYPE_CHECKING:
from .match import FilterContextVars
__all__ = (
"compile",
"CompoundJSONPath",
"findall_async",
"findall",
"finditer_async",
"finditer",
"JSONPatch",
"JSONPatchError",
"JSONPatchTestFailure",
"JSONPath",
"JSONPathEnvironment",
"JSONPathError",
"JSONPathIndexError",
"JSONPathMatch",
"JSONPathNameError",
"JSONPathSyntaxError",
"JSONPathTypeError",
"JSONPointer",
"JSONPointerError",
"JSONPointerIndexError",
"JSONPointerKeyError",
"JSONPointerResolutionError",
"JSONPointerTypeError",
"Lexer",
"NodeList",
"match",
"Parser",
"Projection",
"query",
"Query",
"RelativeJSONPointer",
"RelativeJSONPointerError",
"RelativeJSONPointerIndexError",
"RelativeJSONPointerSyntaxError",
"resolve",
"JSON",
"JSONData",
"JSONScalar",
"UNDEFINED",
)
# For convenience and to delegate to strict or non-strict environments.
DEFAULT_ENV = JSONPathEnvironment()
_STRICT_ENV = JSONPathEnvironment(strict=True)
def compile(path: str, *, strict: bool = False) -> Union[JSONPath, CompoundJSONPath]: # noqa: A001
"""Prepare a path string ready for repeated matching against different data.
Arguments:
path: A JSONPath as a string.
strict: When `True`, compile the path for strict compliance with RFC 9535.
Returns:
A `JSONPath` or `CompoundJSONPath`, ready to match against some data.
Expect a `CompoundJSONPath` if the path string uses the _union_ or
_intersection_ operators.
Raises:
JSONPathSyntaxError: If _path_ is invalid.
JSONPathTypeError: If filter functions are given arguments of an
unacceptable type.
"""
return _STRICT_ENV.compile(path) if strict else DEFAULT_ENV.compile(path)
def findall(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> List[object]:
"""Find all objects in _data_ matching the JSONPath _path_.
If _data_ is a string or a file-like objects, it will be loaded
using `json.loads()` and the default `JSONDecoder`.
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
A list of matched objects. If there are no matches, the list will
be empty.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
_STRICT_ENV.findall(path, data, filter_context=filter_context)
if strict
else DEFAULT_ENV.findall(path, data, filter_context=filter_context)
)
async def findall_async(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> List[object]:
"""Find all objects in _data_ matching the JSONPath _path_.
If _data_ is a string or a file-like objects, it will be loaded
using `json.loads()` and the default `JSONDecoder`.
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
A list of matched objects. If there are no matches, the list will
be empty.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
await _STRICT_ENV.findall_async(path, data, filter_context=filter_context)
if strict
else await DEFAULT_ENV.findall_async(path, data, filter_context=filter_context)
)
def finditer(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> Iterable[JSONPathMatch]:
"""Generate `JSONPathMatch` objects for each match of _path_ in _data_.
If _data_ is a string or a file-like objects, it will be loaded using
`json.loads()` and the default `JSONDecoder`.
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
An iterator yielding `JSONPathMatch` objects for each match.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
_STRICT_ENV.finditer(path, data, filter_context=filter_context)
if strict
else DEFAULT_ENV.finditer(path, data, filter_context=filter_context)
)
async def finditer_async(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> AsyncIterable[JSONPathMatch]:
"""Find all objects in _data_ matching the JSONPath _path_.
If _data_ is a string or a file-like objects, it will be loaded
using `json.loads()` and the default `JSONDecoder`.
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
A list of matched objects. If there are no matches, the list will
be empty.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
await _STRICT_ENV.finditer_async(path, data, filter_context=filter_context)
if strict
else await DEFAULT_ENV.finditer_async(path, data, filter_context=filter_context)
)
def match(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> Union[JSONPathMatch, None]:
"""Return a `JSONPathMatch` instance for the first object found in _data_.
`None` is returned if there are no matches.
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
A `JSONPathMatch` object for the first match, or `None` if there were
no matches.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
_STRICT_ENV.match(path, data, filter_context=filter_context)
if strict
else DEFAULT_ENV.match(path, data, filter_context=filter_context)
)
def query(
path: str,
data: JSONData,
*,
filter_context: Optional[FilterContextVars] = None,
strict: bool = False,
) -> Query:
"""Return a `Query` iterator over matches found by applying _path_ to _data_.
`Query` objects are iterable.
```
for match in jsonpath.query("$.foo..bar", data):
...
```
You can skip and limit results with `Query.skip()` and `Query.limit()`.
```
matches = (
jsonpath.query("$.foo..bar", data)
.skip(5)
.limit(10)
)
for match in matches
...
```
`Query.tail()` will get the last _n_ results.
```
for match in jsonpath.query("$.foo..bar", data).tail(5):
...
```
Get values for each match using `Query.values()`.
```
for obj in jsonpath.query("$.foo..bar", data).limit(5).values():
...
```
Arguments:
path: The JSONPath as a string.
data: A JSON document or Python object implementing the `Sequence`
or `Mapping` interfaces.
filter_context: Arbitrary data made available to filters using
the _filter context_ selector.
strict: When `True`, compile and evaluate with strict compliance with
RFC 9535.
Returns:
A query iterator.
Raises:
JSONPathSyntaxError: If the path is invalid.
JSONPathTypeError: If a filter expression attempts to use types in
an incompatible way.
"""
return (
_STRICT_ENV.query(path, data, filter_context=filter_context)
if strict
else DEFAULT_ENV.query(path, data, filter_context=filter_context)
)
|