22334798432

Committed 23 Feb 2026 06:42PM UTC coverage: 86.956% (-0.02%) from 86.973%

Build # 22334798432

Build Type

push

github

Committed by

web-flow

Commit Message

S3: regenerate test snapshots & parity fixes (#13824)

Coverage Stats

69831 of 80306 relevant lines covered (86.96%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.35

/localstack-core/localstack/services/stepfunctions/asl/jsonata/jsonata.py

from __future__ import annotations

import json
import re
from collections.abc import Callable
from pathlib import Path
from typing import Any, Final

import jpype
import jpype.imports  # noqa # Required for JVM Java class imports

from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
from localstack.services.stepfunctions.packages import jpype_jsonata_package
from localstack.utils.objects import singleton_factory

JSONataExpression = str
VariableReference = str
VariableDeclarations = str


# TODO: move the extraction logic to a formal ANTLR-base parser, as done with legacy
#       Intrinsic Functions in package localstack.services.stepfunctions.asl.antlr
#       with grammars ASLIntrinsicLexer and ASLIntrinsicParser, later used by upstream
#       logics such as in:
#       localstack.services.stepfunctions.asl.parse.intrinsic.preprocessor.Preprocessor
_PATTERN_VARIABLE_REFERENCE = re.compile(
    # 1) Non-capturing branch for JSONata regex literal
    #    /.../ (slash delimited), allowing escaped slashes \/
    r"(?:\/(?:\\.|[^\\/])*\/[a-zA-Z]*)"
    r"|"
    # 2) Non-capturing branch for JSONata string literal:
    #    "..." (double quotes) or '...' (single quotes),
    #    allowing escapes
    r"(?:\"(?:\\.|[^\"\\])*\"|\'(?:\\.|[^\'\\])*\')"
    r"|"
    # 3) Capturing branch for $identifier[.prop…]
    #    Requires at least one identifier character after $, so bare $ (the
    #    JSONata context variable used in filter predicates like [$ = 1]) is
    #    never captured.  $$ is captured but filtered out downstream.
    r"(\$[A-Za-z0-9_$]+(?:\.[A-Za-z0-9_][A-Za-z0-9_$]*)*)"
)

_ILLEGAL_VARIABLE_REFERENCES: Final[set[str]] = {"$", "$$"}
_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR: Final[str] = ":="
_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL: Final[str] = ";"
_EXPRESSION_OPEN_SYMBOL: Final[str] = "("
_EXPRESSION_CLOSE_SYMBOL: Final[str] = ")"


class JSONataException(Exception):
    error: Final[str]
    details: str | None

    def __init__(self, error: str, details: str | None):
        self.error = error
        self.details = details


class _JSONataJVMBridge:
    _java_OBJECT_MAPPER: "com.fasterxml.jackson.databind.ObjectMapper"  # noqa
    _java_JSONATA: "com.dashjoin.jsonata.Jsonata.jsonata"  # noqa

    def __init__(self):
        installer = jpype_jsonata_package.get_installer()
        installer.install()

        from jpype import config as jpype_config

        jpype_config.destroy_jvm = False

        # Limitation: We can only start one JVM instance within LocalStack and using JPype for another purpose
        # (e.g., event-ruler) fails unless we change the way we load/reload the classpath.
        jvm_path = installer.get_java_lib_path()
        jsonata_libs_path = Path(installer.get_installed_dir())
        jsonata_libs_pattern = jsonata_libs_path.joinpath("*")
        jpype.startJVM(jvm_path, classpath=[jsonata_libs_pattern], interrupt=False)

        from com.fasterxml.jackson.databind import ObjectMapper  # noqa
        from com.dashjoin.jsonata.Jsonata import jsonata  # noqa

        self._java_OBJECT_MAPPER = ObjectMapper()
        self._java_JSONATA = jsonata

    @staticmethod
    @singleton_factory
    def get() -> _JSONataJVMBridge:
        return _JSONataJVMBridge()

    def eval_jsonata(self, jsonata_expression: JSONataExpression) -> Any:
        try:
            # Evaluate the JSONata expression with the JVM.
            # TODO: Investigate whether it is worth moving this chain of statements (java_*) to a
            #  Java program to reduce i/o between the JVM and this runtime.
            java_expression = self._java_JSONATA(jsonata_expression)
            java_output = java_expression.evaluate(None)
            java_output_string = self._java_OBJECT_MAPPER.writeValueAsString(java_output)

            # Compute a Python json object from the java string, this is to:
            #  1. Ensure we fully end interactions with the JVM about this value here;
            #  2. The output object may undergo under operations that are not compatible
            #     with jpype objects (such as json.dumps, equality, instanceof, etc.).
            result_str: str = str(java_output_string)
            result_json = json.loads(result_str)

            return result_json
        except Exception as ex:
            raise JSONataException("UNKNOWN", str(ex))


# Lazy initialization of the `eval_jsonata` function pointer.
# This ensures the JVM is only started when JSONata functionality is needed.
_eval_jsonata: Callable[[JSONataExpression], Any] | None = None


def eval_jsonata_expression(jsonata_expression: JSONataExpression) -> Any:
    global _eval_jsonata
    if _eval_jsonata is None:
        # Initialize _eval_jsonata only when invoked for the first time using the Singleton pattern.
        _eval_jsonata = _JSONataJVMBridge.get().eval_jsonata
    return _eval_jsonata(jsonata_expression)


class IllegalJSONataVariableReference(ValueError):
    variable_reference: Final[VariableReference]

    def __init__(self, variable_reference: VariableReference):
        self.variable_reference = variable_reference


def extract_jsonata_variable_references(
    jsonata_expression: JSONataExpression,
) -> set[VariableReference]:
    if not jsonata_expression:
        return set()
    # Extract all recognised patterns.
    all_references: list[Any] = _PATTERN_VARIABLE_REFERENCE.findall(jsonata_expression)
    # Filter non-empty patterns (this includes consumed blocks such as jsonata
    # regular expressions, delimited between non-escaped slashes).
    variable_references: set[VariableReference] = {
        reference for reference in all_references if reference and isinstance(reference, str)
    }
    for variable_reference in variable_references:
        if variable_reference in _ILLEGAL_VARIABLE_REFERENCES:
            raise IllegalJSONataVariableReference(variable_reference=variable_reference)
    return variable_references


def encode_jsonata_variable_declarations(
    bindings: dict[VariableReference, Any],
) -> VariableDeclarations:
    declarations_parts: list[str] = []
    for variable_reference, value in bindings.items():
        if isinstance(value, str):
            value_str_lit = f'"{value}"'
        else:
            value_str_lit = to_json_str(value, separators=(",", ":"))
        declarations_parts.extend(
            [
                variable_reference,
                _VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR,
                value_str_lit,
                _VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL,
            ]
        )
    return "".join(declarations_parts)


def compose_jsonata_expression(
    final_jsonata_expression: JSONataExpression,
    variable_declarations_list: list[VariableDeclarations],
) -> JSONataExpression:
    variable_declarations = "".join(variable_declarations_list)
    expression = "".join(
        [
            _EXPRESSION_OPEN_SYMBOL,
            variable_declarations,
            final_jsonata_expression,
            _EXPRESSION_CLOSE_SYMBOL,
        ]
    )
    return expression

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	import re	1✔
5	from collections.abc import Callable	1✔
6	from pathlib import Path	1✔
7	from typing import Any, Final	1✔
8
9	import jpype	1✔
10	import jpype.imports # noqa # Required for JVM Java class imports	1✔
11
12	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
13	from localstack.services.stepfunctions.packages import jpype_jsonata_package	1✔
14	from localstack.utils.objects import singleton_factory	1✔
15
16	JSONataExpression = str	1✔
17	VariableReference = str	1✔
18	VariableDeclarations = str	1✔
19
20
21	# TODO: move the extraction logic to a formal ANTLR-base parser, as done with legacy
22	# Intrinsic Functions in package localstack.services.stepfunctions.asl.antlr
23	# with grammars ASLIntrinsicLexer and ASLIntrinsicParser, later used by upstream
24	# logics such as in:
25	# localstack.services.stepfunctions.asl.parse.intrinsic.preprocessor.Preprocessor
26	_PATTERN_VARIABLE_REFERENCE = re.compile(	1✔
27	# 1) Non-capturing branch for JSONata regex literal
28	# /.../ (slash delimited), allowing escaped slashes \/
29	r"(?:\/(?:\\.\|[^\\/])\/[a-zA-Z])"
30	r"\|"
31	# 2) Non-capturing branch for JSONata string literal:
32	# "..." (double quotes) or '...' (single quotes),
33	# allowing escapes
34	r"(?:\"(?:\\.\|[^\"\\])\"\|\'(?:\\.\|[^\'\\])\')"
35	r"\|"
36	# 3) Capturing branch for $identifier[.prop…]
37	# Requires at least one identifier character after $, so bare $ (the
38	# JSONata context variable used in filter predicates like [$ = 1]) is
39	# never captured. $$ is captured but filtered out downstream.
40	r"(\$[A-Za-z0-9_$]+(?:\.[A-Za-z0-9_][A-Za-z0-9_$]))"
41	)
42
43	_ILLEGAL_VARIABLE_REFERENCES: Final[set[str]] = {"$", "$$"}	1✔
44	_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR: Final[str] = ":="	1✔
45	_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL: Final[str] = ";"	1✔
46	_EXPRESSION_OPEN_SYMBOL: Final[str] = "("	1✔
47	_EXPRESSION_CLOSE_SYMBOL: Final[str] = ")"	1✔
48
49
50	class JSONataException(Exception):	1✔
51	error: Final[str]	1✔
52	details: str \| None	1✔
53
54	def __init__(self, error: str, details: str \| None):	1✔
55	self.error = error	×
56	self.details = details	×
57
58
59	class _JSONataJVMBridge:	1✔
60	_java_OBJECT_MAPPER: "com.fasterxml.jackson.databind.ObjectMapper" # noqa	1✔
61	_java_JSONATA: "com.dashjoin.jsonata.Jsonata.jsonata" # noqa	1✔
62
63	def __init__(self):	1✔
64	installer = jpype_jsonata_package.get_installer()	1✔
65	installer.install()	1✔
66
67	from jpype import config as jpype_config	1✔
68
69	jpype_config.destroy_jvm = False	1✔
70
71	# Limitation: We can only start one JVM instance within LocalStack and using JPype for another purpose
72	# (e.g., event-ruler) fails unless we change the way we load/reload the classpath.
73	jvm_path = installer.get_java_lib_path()	1✔
74	jsonata_libs_path = Path(installer.get_installed_dir())	1✔
75	jsonata_libs_pattern = jsonata_libs_path.joinpath("*")	1✔
76	jpype.startJVM(jvm_path, classpath=[jsonata_libs_pattern], interrupt=False)	1✔
77
78	from com.fasterxml.jackson.databind import ObjectMapper # noqa	1✔
79	from com.dashjoin.jsonata.Jsonata import jsonata # noqa	1✔
80
81	self._java_OBJECT_MAPPER = ObjectMapper()	1✔
82	self._java_JSONATA = jsonata	1✔
83
84	@staticmethod	1✔
85	@singleton_factory	1✔
86	def get() -> _JSONataJVMBridge:	1✔
87	return _JSONataJVMBridge()	1✔
88
89	def eval_jsonata(self, jsonata_expression: JSONataExpression) -> Any:	1✔
90	try:	1✔
91	# Evaluate the JSONata expression with the JVM.
92	# TODO: Investigate whether it is worth moving this chain of statements (java_*) to a
93	# Java program to reduce i/o between the JVM and this runtime.
94	java_expression = self._java_JSONATA(jsonata_expression)	1✔
95	java_output = java_expression.evaluate(None)	1✔
96	java_output_string = self._java_OBJECT_MAPPER.writeValueAsString(java_output)	1✔
97
98	# Compute a Python json object from the java string, this is to:
99	# 1. Ensure we fully end interactions with the JVM about this value here;
100	# 2. The output object may undergo under operations that are not compatible
101	# with jpype objects (such as json.dumps, equality, instanceof, etc.).
102	result_str: str = str(java_output_string)	1✔
103	result_json = json.loads(result_str)	1✔
104
105	return result_json	1✔
106	except Exception as ex:	×
107	raise JSONataException("UNKNOWN", str(ex))	×
108
109
110	# Lazy initialization of the `eval_jsonata` function pointer.
111	# This ensures the JVM is only started when JSONata functionality is needed.
112	_eval_jsonata: Callable[[JSONataExpression], Any] \| None = None	1✔
113
114
115	def eval_jsonata_expression(jsonata_expression: JSONataExpression) -> Any:	1✔
116	global _eval_jsonata
117	if _eval_jsonata is None:	1✔
118	# Initialize _eval_jsonata only when invoked for the first time using the Singleton pattern.
119	_eval_jsonata = _JSONataJVMBridge.get().eval_jsonata	1✔
120	return _eval_jsonata(jsonata_expression)	1✔
121
122
123	class IllegalJSONataVariableReference(ValueError):	1✔
124	variable_reference: Final[VariableReference]	1✔
125
126	def __init__(self, variable_reference: VariableReference):	1✔
127	self.variable_reference = variable_reference	1✔
128
129
130	def extract_jsonata_variable_references(	1✔
131	jsonata_expression: JSONataExpression,
132	) -> set[VariableReference]:
133	if not jsonata_expression:	1✔
134	return set()	1✔
135	# Extract all recognised patterns.
136	all_references: list[Any] = _PATTERN_VARIABLE_REFERENCE.findall(jsonata_expression)	1✔
137	# Filter non-empty patterns (this includes consumed blocks such as jsonata
138	# regular expressions, delimited between non-escaped slashes).
139	variable_references: set[VariableReference] = {	1✔
140	reference for reference in all_references if reference and isinstance(reference, str)
141	}
142	for variable_reference in variable_references:	1✔
143	if variable_reference in _ILLEGAL_VARIABLE_REFERENCES:	1✔
144	raise IllegalJSONataVariableReference(variable_reference=variable_reference)	1✔
145	return variable_references	1✔
146
147
148	def encode_jsonata_variable_declarations(	1✔
149	bindings: dict[VariableReference, Any],
150	) -> VariableDeclarations:
151	declarations_parts: list[str] = []	1✔
152	for variable_reference, value in bindings.items():	1✔
153	if isinstance(value, str):	1✔
154	value_str_lit = f'"{value}"'	1✔
155	else:
156	value_str_lit = to_json_str(value, separators=(",", ":"))	1✔
157	declarations_parts.extend(	1✔
158	[
159	variable_reference,
160	_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR,
161	value_str_lit,
162	_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL,
163	]
164	)
165	return "".join(declarations_parts)	1✔
166
167
168	def compose_jsonata_expression(	1✔
169	final_jsonata_expression: JSONataExpression,
170	variable_declarations_list: list[VariableDeclarations],
171	) -> JSONataExpression:
172	variable_declarations = "".join(variable_declarations_list)	1✔
173	expression = "".join(	1✔
174	[
175	_EXPRESSION_OPEN_SYMBOL,
176	variable_declarations,
177	final_jsonata_expression,
178	_EXPRESSION_CLOSE_SYMBOL,
179	]
180	)
181	return expression	1✔

localstack / localstack / 22334798432

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous