jq, JMESPath and SQL JSON paths compared

python/jmespath_mini.py

"""jmespath_mini: a small JMESPath interpreter (jmespath.org/specification.html), standard library only.

    from jmespath_mini import search
    search("orders[?city == 'Pune'].amount_paise | sum(@)", data)

It reads an expression the way jmespath.py does, the library behind the AWS CLI's --query option: a lexer, a parser
that orders operators by binding power, and an interpreter over the parsed tree. Where jmespath.py goes beyond the
specification, this follows jmespath.py: <, <=, > and >= also order two strings. Nothing is passed to eval().
"""

import json
import math

__all__ = ["search", "JMESPathError"]


class JMESPathError(ValueError):
    """A syntax error, a function called with the wrong types or number of arguments, or an unknown function."""


# ---- the lexer -----------------------------------------------------------------------------------------------------

_SIMPLE = {".": "dot", "*": "star", "]": "rbracket", ",": "comma", ":": "colon", "@": "current", "(": "lparen", ")": "rparen", "{": "lbrace", "}": "rbrace"}
_IDENT_START = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_")
_IDENT = _IDENT_START | set("0123456789")


def _tokens(text):
    out, i, n = [], 0, len(text)

    def until(delim):
        nonlocal i
        start, buf = i, []
        i += 1
        while i < n and text[i] != delim:
            if text[i] == "\\" and i + 1 < n:
                buf.append(text[i])
                i += 1
            buf.append(text[i])
            i += 1
        if i >= n:
            raise JMESPathError(f"the text that starts at position {start} has no closing {delim}")
        i += 1
        return "".join(buf)

    while i < n:
        c = text[i]
        if c in _SIMPLE:
            out.append((_SIMPLE[c], c))
            i += 1
        elif c in _IDENT_START:
            j = i + 1
            while j < n and text[j] in _IDENT:
                j += 1
            out.append(("name", text[i:j]))
            i = j
        elif c in " \t\n\r":
            i += 1
        elif c == "[":
            nxt = text[i + 1 : i + 2]
            kind = {"]": "flatten", "?": "filter"}.get(nxt, "lbracket")
            out.append((kind, c))
            i += 1 if kind == "lbracket" else 2
        elif c == "'":
            out.append(("literal", until("'").replace("\\'", "'")))  # a raw string: no escapes but \'
        elif c == "`":
            out.append(("literal", _json_literal(until("`").replace("\\`", "`"))))
        elif c == '"':
            try:
                name = json.loads('"' + until('"') + '"')
            except json.JSONDecodeError:
                raise JMESPathError("a quoted name must be a JSON string") from None
            out.append(("quoted", name))  # a quoted name: never a function name
        elif c in "|&":
            double = text[i + 1 : i + 2] == c
            out.append(({"|": "or", "&": "and"}[c] if double else {"|": "pipe", "&": "expref"}[c], c))
            i += 2 if double else 1
        elif c in "<>!=":
            eq = text[i + 1 : i + 2] == "="
            if c == "=" and not eq:
                raise JMESPathError("compare with ==, a single = is not an operator")
            kind = {"<": "lt", ">": "gt", "!": "ne" if eq else "not", "=": "eq"}[c] + ("e" if eq and c in "<>" else "")
            out.append((kind, c))
            i += 2 if eq else 1
        elif c == "-" or c.isdigit() and c.isascii():
            j = i + 1
            while j < n and text[j].isdigit() and text[j].isascii():
                j += 1
            if text[i:j] == "-":
                raise JMESPathError("a minus sign must be followed by digits")
            out.append(("number", int(text[i:j])))
            i = j
        else:
            raise JMESPathError(f"unexpected character {c!r} at position {i}")
    out.append(("eof", None))
    return out


def _json_literal(text):
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        try:
            return json.loads('"' + text.lstrip() + '"')  # old literal syntax for text, which jmespath.py still reads
        except json.JSONDecodeError:
            raise JMESPathError(f"`{text}` is not JSON") from None


# ---- the parser ----------------------------------------------------------------------------------------------------

_POWER = {
    "pipe": 1, "or": 2, "and": 3, "eq": 5, "gt": 5, "lt": 5, "gte": 5, "lte": 5, "ne": 5, "flatten": 9, "star": 20,
    "filter": 21, "dot": 40, "not": 45, "lbrace": 50, "lbracket": 55, "lparen": 60,
}  # every other token binds with 0
_PROJECTION_STOP = 10


class _Parser:
    def __init__(self, text):
        self.tokens, self.i = _tokens(text), 0

    def kind(self, ahead=0):
        return self.tokens[min(self.i + ahead, len(self.tokens) - 1)][0]

    def take(self):
        token = self.tokens[self.i]
        self.i = min(self.i + 1, len(self.tokens) - 1)
        return token

    def expect(self, kind):
        if self.kind() != kind:
            raise JMESPathError(f"expected {kind}, found {self.kind()}")
        self.take()

    def parse(self):
        node = self.expression(0)
        if self.kind() != "eof":
            raise JMESPathError(f"unexpected {self.kind()} after a complete expression")
        return node

    def expression(self, power):
        left = self.nud(self.take())
        while power < _POWER.get(self.kind(), 0):
            left = self.led(self.take(), left)
        return left

    def nud(self, token):
        kind, value = token
        if kind == "literal":
            return ("literal", value)
        if kind in ("name", "quoted"):
            if kind == "quoted" and self.kind() == "lparen":
                raise JMESPathError("a function name cannot be quoted")
            return ("field", value)
        if kind == "star":
            right = ("identity",) if self.kind() == "rbracket" else self.projection_rhs(_POWER["star"])
            return ("value_projection", ("identity",), right)
        if kind == "filter":
            return self.filter_projection(("identity",))
        if kind == "lbrace":
            return self.multi_select_hash()
        if kind == "lparen":
            inner = self.expression(0)
            self.expect("rparen")
            return inner
        if kind == "flatten":
            return ("projection", ("flatten", ("identity",)), self.projection_rhs(_POWER["flatten"]))
        if kind == "not":
            return ("not", self.expression(_POWER["not"]))
        if kind == "lbracket":
            if self.kind() in ("number", "colon"):
                return self.index_or_slice(("identity",))
            if self.kind() == "star" and self.kind(1) == "rbracket":
                self.take(), self.take()
                return ("projection", ("identity",), self.projection_rhs(_POWER["star"]))
            return self.multi_select_list()
        if kind == "current":
            return ("current",)
        if kind == "expref":
            return ("expref", self.expression(0))
        raise JMESPathError(f"unexpected {kind}")

    def led(self, token, left):
        kind = token[0]
        if kind == "dot":
            if self.kind() == "star":
                self.take()
                return ("value_projection", left, self.projection_rhs(_POWER["dot"]))
            return ("subexpression", left, self.dot_rhs(_POWER["dot"]))
        if kind in ("pipe", "or", "and"):
            return (kind, left, self.expression(_POWER[kind]))
        if kind in ("eq", "ne", "lt", "lte", "gt", "gte"):
            return ("compare", kind, left, self.expression(_POWER[kind]))
        if kind == "lparen":
            if left[0] != "field":
                raise JMESPathError("only a function name can be followed by (")
            args = []
            while self.kind() != "rparen":
                args.append(self.expression(0))
                if self.kind() == "comma":
                    self.take()
            self.take()
            return ("function", left[1], args)
        if kind == "filter":
            return self.filter_projection(left)
        if kind == "flatten":
            return ("projection", ("flatten", left), self.projection_rhs(_POWER["flatten"]))
        if kind == "lbracket":
            if self.kind() in ("number", "colon"):
                return self.index_or_slice(left)
            self.expect("star")
            self.expect("rbracket")
            return ("projection", left, self.projection_rhs(_POWER["star"]))
        raise JMESPathError(f"unexpected {kind}")

    def index_or_slice(self, left):
        if self.kind() == "colon" or self.kind(1) == "colon":
            parts, at = [None, None, None], 0
            while self.kind() != "rbracket":
                if self.kind() == "colon":
                    at += 1
                    if at == 3:
                        raise JMESPathError("a slice has at most three parts: [start:stop:step]")
                    self.take()
                elif self.kind() == "number":
                    parts[at] = self.take()[1]
                else:
                    raise JMESPathError("a slice holds only whole numbers")
            self.take()
            return ("projection", ("slice", left, *parts), self.projection_rhs(_POWER["star"]))
        index = self.take()[1]
        self.expect("rbracket")
        return ("index", left, index)

    def filter_projection(self, left):
        condition = self.expression(0)
        self.expect("rbracket")
        right = ("identity",) if self.kind() == "flatten" else self.projection_rhs(_POWER["filter"])
        return ("filter_projection", left, condition, right)

    def projection_rhs(self, power):
        """What a projection applies to each item: nothing, or a chain of .name, [index] and [?filter] steps."""
        if _POWER.get(self.kind(), 0) < _PROJECTION_STOP:
            return ("identity",)
        if self.kind() in ("lbracket", "filter"):
            return self.expression(power)
        if self.kind() == "dot":
            self.take()
            return self.dot_rhs(power)
        raise JMESPathError(f"unexpected {self.kind()} after a projection")

    def dot_rhs(self, power):
        if self.kind() in ("name", "quoted", "star"):
            return self.expression(power)
        if self.kind() == "lbracket":
            self.take()
            return self.multi_select_list()
        if self.kind() == "lbrace":
            self.take()
            return self.multi_select_hash()
        raise JMESPathError("after a dot comes a name, *, [ … ] or { … }")

    def multi_select_list(self):
        items = []
        while True:
            items.append(self.expression(0))
            if self.kind() == "rbracket":
                self.take()
                return ("list", items)
            self.expect("comma")

    def multi_select_hash(self):
        pairs = []
        while True:
            kind, key = self.take()
            if kind not in ("name", "quoted"):
                raise JMESPathError("a multi-select hash holds key: expression pairs")
            self.expect("colon")
            pairs.append((key, self.expression(0)))
            if self.kind() == "rbrace":
                self.take()
                return ("hash", pairs)
            self.expect("comma")


# ---- values --------------------------------------------------------------------------------------------------------


class _ExprRef:
    """&expression, for sort_by(), max_by(), min_by() and map()."""

    def __init__(self, node):
        self.node = node


def _type(v):
    if v is None:
        return "null"
    if isinstance(v, bool):
        return "boolean"
    if isinstance(v, (int, float)):
        return "number"
    if isinstance(v, str):
        return "string"
    if isinstance(v, list):
        return "array"
    if isinstance(v, _ExprRef):
        return "expref"
    return "object"


def _false(v):
    """JMESPath's false values: null, false, "", [] and {} (0 is true)."""
    return v is None or v is False or (isinstance(v, (str, list, dict)) and not v)


def _equal(a, b):
    if _type(a) != _type(b):
        return False
    if isinstance(a, list):
        return len(a) == len(b) and all(_equal(x, y) for x, y in zip(a, b))
    if isinstance(a, dict):
        return a.keys() == b.keys() and all(_equal(a[k], b[k]) for k in a)
    return a == b


# ---- the interpreter -----------------------------------------------------------------------------------------------


def _visit(node, value):
    kind = node[0]
    if kind == "field":
        return value.get(node[1]) if isinstance(value, dict) else None
    if kind == "subexpression":
        return _visit(node[2], _visit(node[1], value))
    if kind == "pipe":
        return _visit(node[2], _visit(node[1], value))
    if kind == "index":
        base = _visit(node[1], value)
        if not isinstance(base, list):
            return None
        i = node[2] + len(base) if node[2] < 0 else node[2]
        return base[i] if 0 <= i < len(base) else None
    if kind == "slice":
        base = _visit(node[1], value)
        if not isinstance(base, list):
            return None
        if node[4] == 0:
            raise JMESPathError("a slice step cannot be 0")
        return base[node[2] : node[3] : node[4]]
    if kind in ("projection", "value_projection", "filter_projection"):
        base = _visit(node[1], value)
        if kind == "value_projection":
            items = list(base.values()) if isinstance(base, dict) else None
        else:
            items = base if isinstance(base, list) else None
        if items is None:
            return None
        if kind == "filter_projection":
            items = [item for item in items if not _false(_visit(node[2], item))]
        right = node[3] if kind == "filter_projection" else node[2]
        results = [_visit(right, item) for item in items]
        return [r for r in results if r is not None]  # a projection leaves out the items whose result is null
    if kind == "flatten":
        base = _visit(node[1], value)
        if not isinstance(base, list):
            return None
        out = []
        for item in base:
            out.extend(item) if isinstance(item, list) else out.append(item)
        return out
    if kind in ("identity", "current"):
        return value
    if kind == "literal":
        return node[1]
    if kind == "list":
        return None if value is None else [_visit(child, value) for child in node[1]]
    if kind == "hash":
        return None if value is None else {key: _visit(child, value) for key, child in node[1]}
    if kind == "or":
        left = _visit(node[1], value)
        return _visit(node[2], value) if _false(left) else left
    if kind == "and":
        left = _visit(node[1], value)
        return left if _false(left) else _visit(node[2], value)
    if kind == "not":
        return _false(_visit(node[1], value))
    if kind == "compare":
        op, left, right = node[1], _visit(node[2], value), _visit(node[3], value)
        if op == "eq":
            return _equal(left, right)
        if op == "ne":
            return not _equal(left, right)
        if not ((_type(left) == _type(right) == "number") or (_type(left) == _type(right) == "string")):
            return None  # ordering compares two numbers, or two strings as jmespath.py allows; anything else is null
        return {"lt": left < right, "lte": left <= right, "gt": left > right, "gte": left >= right}[op]
    if kind == "expref":
        return _ExprRef(node[1])
    if kind == "function":
        return _call(node[1], [_visit(arg, value) for arg in node[2]])
    raise JMESPathError(f"cannot evaluate {kind}")


# ---- the built-in functions (specification section "Built-in Functions") -------------------------------------------

_SIGNATURES = {
    "abs": [["number"]], "avg": [["array-number"]], "ceil": [["number"]], "contains": [["array", "string"], []],
    "ends_with": [["string"], ["string"]], "floor": [["number"]], "join": [["string"], ["array-string"]], "keys": [["object"]],
    "length": [["string", "array", "object"]], "map": [["expref"], ["array"]], "max": [["array-number", "array-string"]],
    "max_by": [["array"], ["expref"]], "merge": [["object"], "..."], "min": [["array-number", "array-string"]],
    "min_by": [["array"], ["expref"]], "not_null": [[], "..."], "reverse": [["array", "string"]],
    "sort": [["array-string", "array-number"]], "sort_by": [["array"], ["expref"]], "starts_with": [["string"], ["string"]],
    "sum": [["array-number"]], "to_array": [[]], "to_number": [[]], "to_string": [[]], "type": [[]], "values": [["object"]],
}


def _check(name, args):
    if name not in _SIGNATURES:
        raise JMESPathError(f"unknown function {name}()")
    params = _SIGNATURES[name]
    variadic = params[-1] == "..."
    fixed = params[:-1] if variadic else params
    if (len(args) < len(fixed)) if variadic else (len(args) != len(fixed)):
        raise JMESPathError(f"{name}() takes {'at least ' if variadic else ''}{len(fixed)} argument(s), {len(args)} given")
    for i, arg in enumerate(args):
        allowed = fixed[min(i, len(fixed) - 1)]
        if not allowed:
            continue
        t = _type(arg)
        if t not in [a.split("-")[0] for a in allowed]:
            raise JMESPathError(f"{name}() does not take {t} as argument {i + 1}")
        item_types = [a.split("-")[1] for a in allowed if a.startswith(t + "-")]
        if t == "array" and item_types and "array" not in allowed and arg:
            want = item_types[0] if len(item_types) == 1 else _type(arg[0])
            if want not in item_types or any(_type(x) != want for x in arg):
                raise JMESPathError(f"{name}() needs an array of {' or '.join(item_types)}s as argument {i + 1}")


def _key(ref, item, allowed, name):
    k = _visit(ref.node, item)
    if _type(k) not in allowed:
        raise JMESPathError(f"{name}() needs the expression to give a {' or '.join(allowed)} for every item")
    return k


def _call(name, args):
    _check(name, args)
    a = args[0] if args else None
    if name == "abs":
        return abs(a)
    if name == "avg":
        return sum(a) / len(a) if a else None
    if name == "ceil":
        return math.ceil(a)
    if name == "floor":
        return math.floor(a)
    if name == "contains":
        if isinstance(a, str):
            return isinstance(args[1], str) and args[1] in a
        return any(_equal(x, args[1]) for x in a)
    if name == "ends_with":
        return a.endswith(args[1])
    if name == "starts_with":
        return a.startswith(args[1])
    if name == "join":
        return a.join(args[1])
    if name == "keys":
        return list(a.keys())
    if name == "values":
        return list(a.values())
    if name == "length":
        return len(a)
    if name == "map":
        return [_visit(a.node, item) for item in args[1]]
    if name in ("max", "min"):
        return (max if name == "max" else min)(a) if a else None
    if name in ("max_by", "min_by"):
        if not a:
            return None
        keys = [_key(args[1], item, ("number", "string"), name) for item in a]
        if len({_type(k) for k in keys}) > 1:
            raise JMESPathError(f"{name}() needs the expression to give only numbers or only strings")
        best = (max if name == "max_by" else min)(range(len(a)), key=lambda i: keys[i])
        return a[best]
    if name == "merge":
        out = {}
        for obj in args:
            out.update(obj)
        return out
    if name == "not_null":
        return next((x for x in args if x is not None), None)
    if name == "reverse":
        return a[::-1]
    if name == "sort":
        return sorted(a)
    if name == "sort_by":
        if not a:
            return a
        first = _type(_visit(args[1].node, a[0]))
        if first not in ("number", "string"):
            raise JMESPathError("sort_by() needs the expression to give numbers or strings")
        return sorted(a, key=lambda item: _key(args[1], item, (first,), name))
    if name == "sum":
        return sum(a)
    if name == "to_array":
        return a if isinstance(a, list) else [a]
    if name == "to_number":
        if _type(a) == "number":
            return a
        if isinstance(a, str):
            for convert in (int, float):
                try:
                    return convert(a)
                except ValueError:
                    pass
        return None
    if name == "to_string":
        return a if isinstance(a, str) else json.dumps(a, separators=(",", ":"))
    if name == "type":
        return _type(a)
    raise JMESPathError(f"unknown function {name}()")


def search(expression, data):
    """The value that a JMESPath expression gives for the data (None for null)."""
    result = _visit(_Parser(expression).parse(), data)
    if isinstance(result, _ExprRef):
        raise JMESPathError("an expression reference (&…) has no JSON value of its own")
    return result

python/jsonpath_mini.py

"""jsonpath_mini: a small JSONPath evaluator that follows RFC 9535, written with the standard library only.

    from jsonpath_mini import query
    for path, value in query("$.books[*].title", catalogue):
        print(path, value)

query() reads the whole query before it looks at the document, so a query that RFC 9535 does not allow raises
JSONPathSyntaxError, with the position of the problem, before anything is selected. Nothing in a query is ever run as
code. The result is the nodelist, a list of (normalized path, value) pairs in the order RFC 9535 gives, with object
members in the order of the document. Section numbers in the comments are those of RFC 9535.
"""

import functools
import re
import unicodedata

MAX_EXACT = 2**53 - 1  # integers in a query stay in the I-JSON range of exact values (section 2.1)
BLANK = " \t\n\r"  # the blank space allowed between the parts of a query (section 2.1.1)
DIGITS = "0123456789"


class JSONPathSyntaxError(ValueError):
    """The text is not a JSONPath query of RFC 9535; .position is where the problem was found."""

    def __init__(self, message, position):
        super().__init__(f"{message} (at position {position})")
        self.position = position


class _Nothing:
    """The special result Nothing (section 2.4.1): the absence of a value, which is not null."""

    def __repr__(self):
        return "Nothing"


NOTHING = _Nothing()


def query(text, document):
    """The nodelist that the query selects from the document: a list of (normalized path, value)."""
    segments = _Parser(text).parse()
    return [(normalized_path(path), value) for path, value in _apply(segments, [((), document)], document)]


# ---- running segments and selectors --------------------------------------------------------------------------------
# A node is (path, value): the path is a tuple of member names and array positions from the root.


def _apply(segments, nodes, root):
    """Applies each segment to every node of the nodelist so far, in order (section 2.1.2)."""
    for descendant, selectors in segments:
        out = []
        for node in nodes:
            for target in _self_and_descendants(node) if descendant else (node,):
                for selector in selectors:
                    out.extend(selector(target, root))
        nodes = out
    return nodes


def _children(node):
    """The member values of an object, or the elements of an array, in order (sections 1.1, 2.3.2)."""
    path, value = node
    if isinstance(value, dict):
        return [(path + (name,), item) for name, item in value.items()]
    if isinstance(value, list):
        return [(path + (i,), item) for i, item in enumerate(value)]
    return []


def _self_and_descendants(node):
    """The node, then its descendants: every node before its descendants, arrays in order (section 2.5.2).
    A list of nodes still to visit, not recursion, so a deeply nested document cannot overflow the stack."""
    stack = [node]
    while stack:
        current = stack.pop()
        yield current
        stack.extend(reversed(_children(current)))


def _name_selector(name):
    def select(node, root):
        path, value = node
        if isinstance(value, dict) and name in value:
            return [(path + (name,), value[name])]
        return []  # nothing is selected from a value that is not an object (section 2.3.1.2)

    return select


def _wildcard_selector(node, root):
    return _children(node)


def _index_selector(index):
    def select(node, root):
        path, value = node
        if isinstance(value, list):
            i = index + len(value) if index < 0 else index  # -1 is the last element (section 2.3.3.2)
            if 0 <= i < len(value):
                return [(path + (i,), value[i])]
        return []

    return select


def slice_positions(length, start, end, step):
    """The positions that start:end:step selects in an array of this length (section 2.3.4.2.2)."""
    step = 1 if step is None else step
    if step == 0:
        return []
    if start is None:
        start = 0 if step > 0 else length - 1
    if end is None:
        end = length if step > 0 else -length - 1

    def normalize(i):
        return i if i >= 0 else length + i

    if step > 0:
        lower = min(max(normalize(start), 0), length)
        upper = min(max(normalize(end), 0), length)
        return list(range(lower, upper, step))
    upper = min(max(normalize(start), -1), length - 1)
    lower = min(max(normalize(end), -1), length - 1)
    return list(range(upper, lower, step))


def _slice_selector(start, end, step):
    def select(node, root):
        path, value = node
        if not isinstance(value, list):
            return []
        return [(path + (i,), value[i]) for i in slice_positions(len(value), start, end, step)]

    return select


def _filter_selector(expression):
    def select(node, root):
        return [child for child in _children(node) if expression.test(child[1], root)]

    return select


# ---- normalized paths (section 2.7) --------------------------------------------------------------------------------

_PATH_ESCAPES = {"\b": "\\b", "\f": "\\f", "\n": "\\n", "\r": "\\r", "\t": "\\t", "'": "\\'", "\\": "\\\\"}


def normalized_path(path):
    """$['books'][0]['title']: single quotes, the escapes of section 2.7, non-negative positions."""
    parts = ["$"]
    for step in path:
        if isinstance(step, int):
            parts.append(f"[{step}]")
        else:
            name = "".join(_PATH_ESCAPES.get(c) or (f"\\u{ord(c):04x}" if ord(c) < 0x20 else c) for c in step)
            parts.append(f"['{name}']")
    return "".join(parts)


# ---- filter expressions (sections 2.3.5 and 2.4) -------------------------------------------------------------------


def _is_number(x):
    return isinstance(x, (int, float)) and not isinstance(x, bool)


def _equal(a, b):
    """== of section 2.3.5.2.2: Nothing equals only Nothing, and values of different types are never equal."""
    if a is NOTHING or b is NOTHING:
        return a is b
    if _is_number(a) and _is_number(b):
        return a == b
    if type(a) is not type(b):
        return False
    if isinstance(a, list):
        return len(a) == len(b) and all(_equal(x, y) for x, y in zip(a, b))
    if isinstance(a, dict):
        return a.keys() == b.keys() and all(_equal(a[k], b[k]) for k in a)
    return a == b


def _less(a, b):
    """< of section 2.3.5.2.2: only two numbers or two strings (by Unicode scalar values) can be ordered."""
    if _is_number(a) and _is_number(b):
        return a < b
    if isinstance(a, str) and isinstance(b, str):
        return a < b
    return False


_COMPARE = {
    "==": _equal,
    "!=": lambda a, b: not _equal(a, b),
    "<": _less,
    "<=": lambda a, b: _less(a, b) or _equal(a, b),
    ">": lambda a, b: _less(b, a),
    ">=": lambda a, b: _less(b, a) or _equal(a, b),
}


class _Literal:
    kind = "value"

    def __init__(self, value):
        self.value = value

    def value_of(self, current, root):
        return self.value


class _Query:
    """@... or $... inside a filter; singular when it can select at most one node (section 2.3.5.1)."""

    kind = "nodes"

    def __init__(self, relative, segments, singular):
        self.relative, self.segments, self.singular = relative, segments, singular

    def nodes(self, current, root):
        return _apply(self.segments, [((), current if self.relative else root)], root)

    def value_of(self, current, root):
        found = self.nodes(current, root)
        return found[0][1] if found else NOTHING

    def test(self, current, root):
        return bool(self.nodes(current, root))  # an existence test (section 2.3.5.2.1)


def _length(v):
    if isinstance(v, (str, list, dict)):
        return len(v)  # a Python str has one item for each Unicode scalar value
    return NOTHING


def _regex_test(how):
    def run(text, pattern):
        if not isinstance(text, str) or not isinstance(pattern, str):
            return False
        compiled = _iregexp(pattern)
        return compiled is not None and (compiled.fullmatch(text) if how == "match" else compiled.search(text)) is not None

    return run


def _value(nodes):
    return nodes[0][1] if len(nodes) == 1 else NOTHING


# name: (the declared types of the parameters, the declared result type, the function) (sections 2.4.4 to 2.4.8)
FUNCTIONS = {
    "length": (("value",), "value", _length),
    "count": (("nodes",), "value", len),
    "match": (("value", "value"), "logical", _regex_test("match")),
    "search": (("value", "value"), "logical", _regex_test("search")),
    "value": (("nodes",), "value", _value),
}


class _Call:
    def __init__(self, name, args):
        self.name, self.args = name, args
        self.params, self.kind, self.function = FUNCTIONS[name]

    def _run(self, current, root):
        values = []
        for param, arg in zip(self.params, self.args):
            values.append(arg.nodes(current, root) if param == "nodes" else arg.value_of(current, root))
        return self.function(*values)

    def value_of(self, current, root):
        return self._run(current, root)

    def test(self, current, root):
        result = self._run(current, root)
        return bool(result)  # LogicalType, or a NodesType result converted (section 2.4.2)


class _Compare:
    def __init__(self, left, op, right):
        self.left, self.op, self.right = left, op, right

    def test(self, current, root):
        return _COMPARE[self.op](self.left.value_of(current, root), self.right.value_of(current, root))


class _Not:
    def __init__(self, inner):
        self.inner = inner

    def test(self, current, root):
        return not self.inner.test(current, root)


class _And:
    def __init__(self, left, right):
        self.left, self.right = left, right

    def test(self, current, root):
        return self.left.test(current, root) and self.right.test(current, root)


class _Or:
    def __init__(self, left, right):
        self.left, self.right = left, right

    def test(self, current, root):
        return self.left.test(current, root) or self.right.test(current, root)


# ---- I-Regexp (RFC 9485), translated to a Python regular expression -------------------------------------------------

_NOT_NORMAL = set("()*+.?[\\]{|}")  # the characters NormalChar leaves out
_SINGLE_ESCAPES = set("()*+-.?[\\]^nrt{|}")  # what may follow a backslash (SingleCharEsc)
_CATEGORIES = {"L": "lmotu", "M": "cen", "N": "dlo", "P": "cdefios", "Z": "lps", "S": "ckmo", "C": "cfno"}


@functools.lru_cache(maxsize=None)
def _category_ranges(prop, complement):
    """The code points of a Unicode general category, as ranges for a character class (\\p{..}, \\P{..})."""
    ranges, start = [], None
    for cp in range(0x110000 + 1):
        inside = cp <= 0x10FFFF and not 0xD800 <= cp <= 0xDFFF
        if inside:
            inside = unicodedata.category(chr(cp)).startswith(prop) != complement
        if inside and start is None:
            start = cp
        elif not inside and start is not None:
            ranges.append(f"\\U{start:08x}-\\U{cp - 1:08x}" if cp - 1 > start else f"\\U{start:08x}")
            start = None
    return "".join(ranges)


class _IRegexp:
    """A recursive-descent reader of the I-Regexp grammar (RFC 9485 section 3) that writes Python's syntax."""

    def __init__(self, text):
        self.text, self.pos = text, 0

    def peek(self):
        return self.text[self.pos] if self.pos < len(self.text) else ""

    def take(self):
        c = self.peek()
        if not c:
            raise ValueError("unexpected end")
        self.pos += 1
        return c

    def regexp(self):
        branches = [self.branch()]
        while self.peek() == "|":
            self.pos += 1
            branches.append(self.branch())
        return "|".join(branches)

    def branch(self):
        out = []
        while self.peek() and self.peek() not in "|)":
            atom = self.atom()
            out.append(atom + self.quantifier())
        return "".join(out)

    def quantifier(self):
        c = self.peek()
        if c in ("*", "+", "?") and c:
            self.pos += 1
            return c
        if c != "{":
            return ""
        m = re.compile(r"\{([0-9]+)(,([0-9]*))?\}").match(self.text, self.pos)
        if not m or (m.group(3) and int(m.group(3)) < int(m.group(1))):
            raise ValueError("bad quantifier")
        self.pos = m.end()
        return m.group(0)

    def atom(self):
        c = self.take()
        if c == "(":
            inner = self.regexp()
            if self.take() != ")":
                raise ValueError("missing )")
            return f"(?:{inner})"
        if c == ".":
            return "[^\\n\\r]"  # RFC 9485 section 5.3: the dot matches anything but a line feed or a return
        if c == "[":
            return self.char_class()
        if c == "\\":
            return self.escape(in_class=False)
        if c in _NOT_NORMAL:
            raise ValueError(f"{c} is not allowed here")
        if c in "^$":
            return c  # NormalChar in RFC 9485's grammar; kept as anchors, as section 5.3's translation does
        return re.escape(c)

    def escape(self, in_class):
        c = self.take()
        if c in ("p", "P"):
            if self.take() != "{":
                raise ValueError("expected {")
            end = self.text.find("}", self.pos)
            prop = self.text[self.pos : end] if end >= 0 else ""
            if not (len(prop) in (1, 2) and prop[0] in _CATEGORIES and (len(prop) == 1 or prop[1] in _CATEGORIES[prop[0]])):
                raise ValueError("unknown category")
            self.pos = end + 1
            ranges = _category_ranges(prop, c == "P")
            return ranges if in_class else f"[{ranges}]"
        if c not in _SINGLE_ESCAPES:
            raise ValueError(f"\\{c} is not an escape of I-Regexp")
        return {"n": "\\n", "r": "\\r", "t": "\\t"}.get(c) or re.escape(c)

    def class_char(self):
        """One CCchar: a character, or a single-character escape; returns (its text for Python, the character)."""
        c = self.take()
        if c == "\\":
            nxt = self.peek()
            if nxt in ("p", "P"):
                return self.escape(in_class=True), None
            text = self.escape(in_class=True)
            return text, {"n": "\n", "r": "\r", "t": "\t"}.get(nxt, nxt)
        if c in "-[]":
            raise ValueError(f"{c} must be escaped in a class")
        return re.escape(c), c

    def char_class(self):
        negated = self.peek() == "^"
        if negated:
            self.pos += 1
        items = []
        if self.peek() == "-":
            self.pos += 1
            items.append("\\-")
        while self.peek() and self.peek() != "]":
            if self.peek() == "-" and self.text[self.pos + 1 : self.pos + 2] == "]":
                self.pos += 1
                items.append("\\-")
                break
            text, char = self.class_char()
            if self.peek() == "-" and self.text[self.pos + 1 : self.pos + 2] not in ("]", ""):
                if char is None:
                    raise ValueError("a category cannot start a range")
                self.pos += 1
                end_text, end_char = self.class_char()
                if end_char is None or ord(end_char) < ord(char):
                    raise ValueError("bad range")
                items.append(f"{text}-{end_text}")
            else:
                items.append(text)
        if self.take() != "]" or not items:
            raise ValueError("bad class")
        return f"[{'^' if negated else ''}{''.join(items)}]"


@functools.lru_cache(maxsize=256)
def _iregexp(pattern):
    """The compiled Python form of an I-Regexp, or None when the pattern is not one (match() is then false)."""
    try:
        reader = _IRegexp(pattern)
        translated = reader.regexp()
        if reader.pos != len(pattern):
            return None
        return re.compile(translated)
    except (ValueError, re.error):
        return None


# ---- the parser (the ABNF of sections 2.1 to 2.5) -----------------------------------------------------------------


class _Parser:
    def __init__(self, text):
        self.text, self.pos = text, 0

    def fail(self, message, at=None):
        raise JSONPathSyntaxError(message, self.pos if at is None else at)

    def peek(self, s):
        return self.text.startswith(s, self.pos)

    def eat(self, s):
        if self.text.startswith(s, self.pos):
            self.pos += len(s)
            return True
        return False

    def blank(self):
        while self.pos < len(self.text) and self.text[self.pos] in BLANK:
            self.pos += 1

    def parse(self):
        if not self.eat("$"):
            self.fail("a query starts with $")
        segments, _ = self.segments()
        if self.pos != len(self.text):
            self.fail("unexpected text")
        return segments

    # -- segments ---------------------------------------------------------------------------------------------------

    def segments(self):
        """*(S segment), and whether the segments keep the query singular (section 2.3.5.1)."""
        segments, singular = [], True
        while True:
            before = self.pos
            self.blank()
            if self.eat(".."):  # a descendant segment: ..[…], ..* or ..name (section 2.5.2)
                selectors = self.bracketed()[0] if self.peek("[") else [self.dotted_name_or_wildcard()[0]]
                segments.append((True, selectors))
                singular = False
            elif self.eat("."):  # .* or .name, shorthand for [*] and ['name'] (section 2.5.1)
                selector, is_name = self.dotted_name_or_wildcard()
                segments.append((False, [selector]))
                singular = singular and is_name
            elif self.peek("["):
                selectors, one_simple = self.bracketed()
                segments.append((False, selectors))
                singular = singular and one_simple
            else:
                self.pos = before  # blank space with no segment after it belongs to what follows
                return segments, singular

    def dotted_name_or_wildcard(self):
        if self.eat("*"):
            return _wildcard_selector, False
        start = self.pos
        if self.pos < len(self.text) and _name_first(self.text[self.pos]):
            self.pos += 1
            while self.pos < len(self.text) and (_name_first(self.text[self.pos]) or self.text[self.pos] in DIGITS):
                self.pos += 1
            return _name_selector(self.text[start : self.pos]), True
        self.fail("expected a member name or * after the dot")

    def bracketed(self):
        """[ selector, selector … ]: the selectors, and whether it is one name or one index without blank space."""
        self.pos += 1  # [
        selectors, simple = [], False
        tight = self.pos < len(self.text) and self.text[self.pos] not in BLANK
        while True:
            self.blank()
            selector, simple = self.selector()
            selectors.append(selector)
            end_tight = self.pos < len(self.text) and self.text[self.pos] in "],"
            self.blank()
            if self.eat("]"):
                break
            if not self.eat(","):
                self.fail("expected , or ]")
        return selectors, len(selectors) == 1 and simple and tight and end_tight

    def selector(self):
        """One selector, and whether it is a name or an index (the selectors a singular query may use)."""
        c = self.text[self.pos : self.pos + 1]
        if c in ("'", '"'):
            return _name_selector(self.string()), True
        if c == "*":
            self.pos += 1
            return _wildcard_selector, False
        if c == "?":
            self.pos += 1
            self.blank()
            return _filter_selector(self.logical_or()), False
        if c == "(":
            self.fail("(…) is a script expression of the 2007 article, which RFC 9535 does not have")
        start = self.integer()
        after = self.pos
        self.blank()
        if not self.peek(":"):
            self.pos = after
            if start is None:
                self.fail("expected a selector")
            return _index_selector(start), True
        self.pos += 1
        self.blank()
        end = self.integer()
        if end is not None:
            after = self.pos
            self.blank()
            if not self.peek(":"):
                self.pos = after
        step = None
        if self.eat(":"):
            after = self.pos
            self.blank()
            step = self.integer()
            if step is None:
                self.pos = after
        return _slice_selector(start, end, step), False

    def integer(self):
        """An integer of the query (section 2.3.3.1): no leading zeros, no -0, inside the I-JSON range; or None."""
        m = re.compile(r"-?[0-9]+").match(self.text, self.pos)
        if not m:
            return None
        digits = m.group().lstrip("-")
        if (digits.startswith("0") and len(digits) > 1) or m.group() == "-0":
            self.fail("integers have no leading zeros, and -0 is not one")
        value = int(m.group())
        if abs(value) > MAX_EXACT:
            self.fail("the integer is outside the range -(2^53 - 1) to 2^53 - 1")
        self.pos = m.end()
        return value

    def string(self):
        """A string literal in single or double quotes, with the escapes of section 2.3.1.1."""
        quote = self.text[self.pos]
        self.pos += 1
        out = []
        while True:
            if self.pos >= len(self.text):
                self.fail("the string has no closing quote")
            c = self.text[self.pos]
            self.pos += 1
            if c == quote:
                return "".join(out)
            if c == "\\":
                out.append(self.string_escape(quote))
            elif ord(c) < 0x20:
                self.fail("a control character must be escaped", self.pos - 1)
            else:
                out.append(c)

    def string_escape(self, quote):
        c = self.text[self.pos : self.pos + 1]
        self.pos += 1
        simple = {"b": "\b", "f": "\f", "n": "\n", "r": "\r", "t": "\t", "/": "/", "\\": "\\", quote: quote}
        if c in simple:
            return simple[c]
        if c != "u":
            self.fail(f"\\{c} is not an escape in a {'single' if quote == chr(39) else 'double'}-quoted string", self.pos - 2)
        code = self.hex4()
        if 0xDC00 <= code <= 0xDFFF:
            self.fail("a low surrogate needs a high surrogate before it")
        if 0xD800 <= code <= 0xDBFF:
            if not self.eat("\\u"):
                self.fail("a high surrogate needs a low surrogate after it")
            low = self.hex4()
            if not 0xDC00 <= low <= 0xDFFF:
                self.fail("a high surrogate needs a low surrogate after it")
            return chr(0x10000 + ((code - 0xD800) << 10) + (low - 0xDC00))
        return chr(code)

    def hex4(self):
        digits = self.text[self.pos : self.pos + 4]
        if len(digits) != 4 or any(d not in "0123456789abcdefABCDEF" for d in digits):
            self.fail("\\u needs four hexadecimal digits")
        self.pos += 4
        return int(digits, 16)

    # -- logical expressions ------------------------------------------------------------------------------------------

    def logical_or(self):
        left = self.logical_and()
        while True:
            before = self.pos
            self.blank()
            if not self.eat("||"):
                self.pos = before
                return left
            self.blank()
            left = _Or(left, self.logical_and())

    def logical_and(self):
        left = self.basic()
        while True:
            before = self.pos
            self.blank()
            if not self.eat("&&"):
                self.pos = before
                return left
            self.blank()
            left = _And(left, self.basic())

    def paren(self):
        self.pos += 1  # (
        self.blank()
        inner = self.logical_or()
        self.blank()
        if not self.eat(")"):
            self.fail("expected )")
        return inner

    def basic(self):
        if self.eat("!"):
            self.blank()
            if self.peek("("):
                return _Not(self.paren())
            at = self.pos
            operand = self.operand()
            self.no_comparison_after("! applies to a test or to a parenthesised expression; write !(a == b)")
            return _Not(self.as_test(operand, at))
        if self.peek("("):
            return self.paren()
        at = self.pos
        left = self.operand()
        before = self.pos
        self.blank()
        op = self.comparison_op()
        if op is None:
            self.pos = before
            return self.as_test(left, at)
        self.check_comparable(left, at)
        self.blank()
        right_at = self.pos
        right = self.operand()
        self.check_comparable(right, right_at)
        self.no_comparison_after("a comparison has one operator; join comparisons with && or ||")
        return _Compare(left, op, right)

    def comparison_op(self):
        for op in ("==", "!=", "<=", ">=", "<", ">"):
            if self.eat(op):
                return op
        return None

    def no_comparison_after(self, message):
        before = self.pos
        self.blank()
        if self.comparison_op() is not None:
            self.fail(message)
        self.pos = before

    def operand(self):
        """A literal, a query that starts with @ or $, or a function expression."""
        c = self.text[self.pos : self.pos + 1]
        if c in ("@", "$"):
            self.pos += 1
            segments, singular = self.segments()
            return _Query(c == "@", segments, singular)
        if c in ("'", '"'):
            return _Literal(self.string())
        if c and c in "-" + DIGITS:
            return _Literal(self.number())
        m = re.compile(r"[a-z][a-z0-9_]*").match(self.text, self.pos)
        if not m:
            self.fail("expected a query, a literal or a function")
        name = m.group()
        if name in ("true", "false", "null") and not self.text.startswith("(", m.end()):
            self.pos = m.end()
            return _Literal({"true": True, "false": False, "null": None}[name])
        if not self.text.startswith("(", m.end()):
            self.fail(f"{name} is not a literal; a function name is followed by (")
        if name not in FUNCTIONS:
            self.fail(f"{name}() is not one of the functions length, count, match, search and value")
        at = self.pos
        self.pos = m.end() + 1
        self.blank()
        args = []
        if not self.peek(")"):
            while True:
                args.append(self.argument())
                self.blank()
                if self.eat(")"):
                    break
                if not self.eat(","):
                    self.fail("expected , or )")
                self.blank()
        else:
            self.pos += 1
        return self.check_call(name, args, at)

    def argument(self):
        """A function argument: a literal, a query or a function alone, else a logical expression."""
        at = self.pos
        if not self.peek("(") and not self.peek("!"):
            operand = self.operand()
            after = self.pos
            self.blank()
            if self.peek(",") or self.peek(")"):
                self.pos = after
                return (operand, at)
            self.pos = at
        return (self.logical_or(), at)

    def check_call(self, name, args, at):
        """Well-typedness (section 2.4.3): every argument fits the declared type of its parameter."""
        params = FUNCTIONS[name][0]
        if len(args) != len(params):
            self.fail(f"{name}() takes {len(params)} argument{'s' if len(params) > 1 else ''}", at)
        checked = []
        for param, (arg, arg_at) in zip(params, args):
            if param == "value":
                ok = isinstance(arg, _Literal) or (isinstance(arg, _Query) and arg.singular) or (isinstance(arg, _Call) and arg.kind == "value")
                why = "a literal, a singular query or a function that gives a value"
            else:
                ok = isinstance(arg, _Query) or (isinstance(arg, _Call) and arg.kind == "nodes")
                why = "a query"
            if not ok:
                self.fail(f"the argument of {name}() must be {why}", arg_at)
            checked.append(arg)
        return _Call(name, checked)

    def check_comparable(self, operand, at):
        if isinstance(operand, _Literal) or (isinstance(operand, _Call) and operand.kind == "value"):
            return
        if isinstance(operand, _Query) and operand.singular:
            return
        self.fail("only literals, singular queries and functions that give a value can be compared", at)

    def as_test(self, operand, at):
        if isinstance(operand, _Query) or (isinstance(operand, _Call) and operand.kind in ("logical", "nodes")):
            return operand
        if isinstance(operand, _Literal):
            self.fail("a literal on its own is not a test; compare it with something", at)
        self.fail(f"{operand.name}() gives a value, which is not a test; compare it with something", at)

    def number(self):
        """A number literal (section 2.3.5.1): an integer or -0, an optional fraction and exponent."""
        m = re.compile(r"(-?(?:0|[1-9][0-9]*))(\.[0-9]+)?([eE][-+]?[0-9]+)?").match(self.text, self.pos)
        if not m or (self.text[m.end() : m.end() + 1] in DIGITS and self.text[m.end() : m.end() + 1]):
            self.fail("not a number")
        self.pos = m.end()
        return float(m.group()) if m.group(2) or m.group(3) else int(m.group(1))


def _name_first(c):
    """ALPHA, _, or any character from U+0080 on (section 2.5.1.1)."""
    return c == "_" or ("a" <= c <= "z") or ("A" <= c <= "Z") or ord(c) >= 0x80