christiangeorgelucas/jmespath-tools

import json

import jmespath
from jmespath import exceptions as jmespath_exceptions

from gen.messages_pb2 import ParseRequest, ParseResult
from gen.axiom_context import AxiomContext


def parse(ax: AxiomContext, input: ParseRequest) -> ParseResult:
    """Parses a JMESPath expression into its abstract syntax tree, without
    evaluating it against any data -- useful for tooling that introspects,
    analyzes, or programmatically builds expressions (e.g. a visual query
    builder, or checking which fields/functions an expression touches). The
    AST is jmespath.py's own parsed tree: a nested structure of node "type"
    (field, subexpression, index, slice, projection, filter_projection,
    function_expression, comparator, pipe, or_expression, and_expression,
    not_expression, literal, multi_select_list, multi_select_hash, expref,
    current, ...), "children", and, for identifier/literal/function nodes, a
    "value".
    """
    try:
        try:
            compiled = jmespath.compile(input.expression)
        except jmespath_exceptions.JMESPathError as e:
            return ParseResult(valid=False, error=str(e))

        ast_json = json.dumps(compiled.parsed, sort_keys=True)
        return ParseResult(valid=True, ast=ast_json)
    except Exception as e:  # never leak a raw traceback for any input
        return ParseResult(valid=False, error=f"unexpected error: {e}")

import jmespath
from jmespath import exceptions as jmespath_exceptions

from gen.messages_pb2 import ProjectPathsRequest, ProjectPathsResult
from gen.axiom_context import AxiomContext
from nodes._common import parse_json_document, dump_json_value


def project_paths(ax: AxiomContext, input: ProjectPathsRequest) -> ProjectPathsResult:
    """Evaluates several independent JMESPath expressions against the same
    document and assembles the results into one flat JSON object keyed by
    name -- the structured equivalent of JMESPath's multiselect-hash syntax
    ("{a: expr_a, b: expr_b}"), without requiring the caller to hand-author
    that syntax (whose colons and nested quoting are awkward to embed
    correctly inside a JSON string value). All field names must be unique
    and every field's expression must parse and evaluate cleanly, or the
    whole request fails with a structured error naming the offending field.
    """
    try:
        try:
            document = parse_json_document(input.json)
        except ValueError as e:
            return ProjectPathsResult(error=str(e))

        seen_names = set()
        assembled = {}
        for field in input.fields:
            if field.name in seen_names:
                return ProjectPathsResult(
                    error=f"duplicate field name: {field.name!r}"
                )
            seen_names.add(field.name)

            try:
                compiled = jmespath.compile(field.expression)
                value = compiled.search(document)
            except jmespath_exceptions.JMESPathError as e:
                return ProjectPathsResult(
                    error=f"field {field.name!r}: {e}"
                )
            assembled[field.name] = value

        return ProjectPathsResult(
            json=dump_json_value(assembled, input.pretty, input.indent)
        )
    except Exception as e:  # never leak a raw traceback for any input
        return ProjectPathsResult(error=f"unexpected error: {e}")

import jmespath
from jmespath import exceptions as jmespath_exceptions

from gen.messages_pb2 import SearchRequest, SearchResult
from gen.axiom_context import AxiomContext
from nodes._common import parse_json_document, dump_json_value


def search(ax: AxiomContext, input: SearchRequest) -> SearchResult:
    """Evaluates a JMESPath expression (https://jmespath.org) against a JSON
    document and returns the single result value the expression produces,
    using jmespath.py -- the reference implementation AWS CLI's --query and
    Ansible's json_query filter are built on. JMESPath makes no distinction
    between "the path is absent" and "the value is JSON null" -- both
    evaluate to null, per the language spec.
    """
    try:
        try:
            document = parse_json_document(input.json)
        except ValueError as e:
            return SearchResult(error=str(e))

        try:
            compiled = jmespath.compile(input.expression)
        except jmespath_exceptions.JMESPathError as e:
            return SearchResult(error=f"invalid JMESPath expression: {e}")

        try:
            result = compiled.search(document)
        except jmespath_exceptions.JMESPathError as e:
            return SearchResult(error=f"expression evaluation failed: {e}")

        return SearchResult(json=dump_json_value(result, input.pretty, input.indent))
    except Exception as e:  # never leak a raw traceback for any input
        return SearchResult(error=f"unexpected error: {e}")

import jmespath
from jmespath import exceptions as jmespath_exceptions

from gen.messages_pb2 import SearchBatchRequest, SearchBatchResult, BatchItem
from gen.axiom_context import AxiomContext
from nodes._common import parse_json_document, dump_json_value


def search_batch(ax: AxiomContext, input: SearchBatchRequest) -> SearchBatchResult:
    """Evaluates one JMESPath expression against each of several independent
    JSON documents, with per-document error isolation -- distinct from
    running one Search against a single document that happens to contain an
    array (there, one malformed element fails the whole request; here, each
    document succeeds or fails on its own and every outcome is reported).
    The expression is compiled exactly once and reused across all documents.
    """
    try:
        try:
            compiled = jmespath.compile(input.expression)
        except jmespath_exceptions.JMESPathError as e:
            return SearchBatchResult(error=f"invalid JMESPath expression: {e}")

        items = []
        for index, doc_json in enumerate(input.json_documents):
            try:
                document = parse_json_document(doc_json)
            except ValueError as e:
                items.append(BatchItem(index=index, error=str(e)))
                continue

            try:
                result = compiled.search(document)
            except jmespath_exceptions.JMESPathError as e:
                items.append(
                    BatchItem(index=index, error=f"expression evaluation failed: {e}")
                )
                continue

            items.append(
                BatchItem(
                    index=index,
                    json=dump_json_value(result, input.pretty, input.indent),
                )
            )

        return SearchBatchResult(items=items)
    except Exception as e:  # never leak a raw traceback for any input
        return SearchBatchResult(error=f"unexpected error: {e}")

import jmespath
from jmespath import exceptions as jmespath_exceptions

from gen.messages_pb2 import ValidateRequest, ValidateResult
from gen.axiom_context import AxiomContext


def validate(ax: AxiomContext, input: ValidateRequest) -> ValidateResult:
    """Checks whether a JMESPath expression is syntactically valid and
    compiles successfully, without running it against any data -- useful
    for checking an expression before spending an invocation evaluating it
    against real data.
    """
    try:
        jmespath.compile(input.expression)
        return ValidateResult(valid=True)
    except jmespath_exceptions.JMESPathError as e:
        return ValidateResult(valid=False, error=str(e))
    except Exception as e:  # never leak a raw traceback for any input
        return ValidateResult(valid=False, error=f"unexpected error: {e}")