65ebfa7b-38ab-49bb-b71e-44800576a11c

Committed 14 Mar 2025 12:37AM UTC coverage: 86.954% (+0.02%) from 86.93%

Build # 65ebfa7b-38ab-49bb-b71e-44800576a11c

Build Type

push

circleci

Committed by

web-flow

Commit Message

Fix apigw input path formatting (#12379)

Coverage Stats

43 of 49 new or added lines in 2 files covered. (87.76%)

103 existing lines in 16 files now uncovered.

62313 of 71662 relevant lines covered (86.95%)

0.87 hits per line

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

89.03

/localstack-core/localstack/services/apigateway/next_gen/execute_api/template_mapping.py

# > In API Gateway, an API's method request or response can take a payload in a different format from the integration
# request or response.
#
# You can transform your data to:
# - Match the payload to an API-specified format.
# - Override an API's request and response parameters and status codes.
# - Return client selected response headers.
# - Associate path parameters, query string parameters, or header parameters in the method request of HTTP proxy
#       or AWS service proxy. TODO: this is from the documentation. Can we use requestOverides for proxy integrations?
# - Select which data to send using integration with AWS services, such as Amazon DynamoDB or Lambda functions,
#       or HTTP endpoints.
#
# You can use mapping templates to transform your data. A mapping template is a script expressed in Velocity Template
# Language (VTL) and applied to the payload using JSONPath .
#
# https://docs.aws.amazon.com/apigateway/latest/developerguide/models-mappings.html
import base64
import copy
import json
import logging
from typing import Any, TypedDict
from urllib.parse import quote_plus, unquote_plus

import airspeed
from airspeed.operators import dict_to_string

from localstack import config
from localstack.services.apigateway.next_gen.execute_api.variables import (
    ContextVariables,
    ContextVarsRequestOverride,
    ContextVarsResponseOverride,
)
from localstack.utils.aws.templating import APIGW_SOURCE, VelocityUtil, VtlTemplate
from localstack.utils.json import extract_jsonpath, json_safe

LOG = logging.getLogger(__name__)


class MappingTemplateParams(TypedDict, total=False):
    path: dict[str, str]
    querystring: dict[str, str]
    header: dict[str, str]


class MappingTemplateInput(TypedDict, total=False):
    body: str
    params: MappingTemplateParams


class MappingTemplateVariables(TypedDict, total=False):
    context: ContextVariables
    input: MappingTemplateInput
    stageVariables: dict[str, str]


def cast_to_vtl_object(value):
    if isinstance(value, dict):
        return VTLMap(value)
    if isinstance(value, list):
        return [cast_to_vtl_object(item) for item in value]
    return value


def cast_to_vtl_json_object(value: Any) -> Any:
    if isinstance(value, dict):
        return VTLJsonDict(value)
    if isinstance(value, list):
        return VTLJsonList(value)
    return value


class VTLMap(dict):
    """Overrides __str__ of python dict (and all child dict) to return a Java like string representation"""

    # TODO apply this class more generally through the template mappings

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.update(*args, **kwargs)

    @staticmethod
    def cast_factory(value: Any) -> Any:
        return cast_to_vtl_object(value)

    def update(self, *args, **kwargs):
        for k, v in self.items():
            self[k] = self.cast_factory(v)

    def __str__(self) -> str:
        return dict_to_string(self)


class VTLJsonList(list):
    """Some VTL List behave differently when being represented as string and everything
    inside will be represented as a json string

    Example: $input.path('$').b // Where path is {"a": 1, "b": [{"c": 5}]}
    Results: '[{"c":5}]' // Where everything inside the list is a valid json object
    """

    def __init__(self, *args):
        super(VTLJsonList, self).__init__(*args)
        for idx, item in enumerate(self):
            self[idx] = cast_to_vtl_json_object(item)

    def __str__(self):
        if isinstance(self, list):
            return json.dumps(self, separators=(",", ":"))


class VTLJsonDict(VTLMap):
    """Some VTL Map behave differently when being represented as string and a list
    encountered in the dictionary will be represented as a json string

    Example: $input.path('$') // Where path is {"a": 1, "b": [{"c": 5}]}
    Results: '{a=1, b=[{"c":5}]}' // Where everything inside the list is a valid json object
    """

    @staticmethod
    def cast_factory(value: Any) -> Any:
        return cast_to_vtl_json_object(value)


class AttributeDict(dict):
    """
    Wrapper returned by VelocityUtilApiGateway.parseJson to allow access to dict values as attributes (dot notation),
    e.g.: $util.parseJson('$.foo').bar
    """

    def __init__(self, *args, **kwargs):
        super(AttributeDict, self).__init__(*args, **kwargs)
        for key, value in self.items():
            if isinstance(value, dict):
                self[key] = AttributeDict(value)

    def __getattr__(self, name):
        if name in self:
            return self[name]
        raise AttributeError(f"'AttributeDict' object has no attribute '{name}'")

    def __setattr__(self, name, value):
        self[name] = value

    def __delattr__(self, name):
        if name in self:
            del self[name]
        else:
            raise AttributeError(f"'AttributeDict' object has no attribute '{name}'")


class VelocityUtilApiGateway(VelocityUtil):
    """
    Simple class to mimic the behavior of variable '$util' in AWS API Gateway integration
    velocity templates.
    See: https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
    """

    def base64Encode(self, s):
        if not isinstance(s, str):
            s = json.dumps(s)
        encoded_str = s.encode(config.DEFAULT_ENCODING)
        encoded_b64_str = base64.b64encode(encoded_str)
        return encoded_b64_str.decode(config.DEFAULT_ENCODING)

    def base64Decode(self, s):
        if not isinstance(s, str):
            s = json.dumps(s)
        return base64.b64decode(s)

    def toJson(self, obj):
        return obj and json.dumps(obj)

    def urlEncode(self, s):
        return quote_plus(s)

    def urlDecode(self, s):
        return unquote_plus(s)

    def escapeJavaScript(self, obj: Any) -> str:
        """
        Converts the given object to a string and escapes any regular single quotes (') into escaped ones (\').
        JSON dumps will escape the single quotes.
        https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
        """
        if obj is None:
            return "null"
        if isinstance(obj, str):
            # empty string escapes to empty object
            if len(obj.strip()) == 0:
                return "{}"
            return json.dumps(obj)[1:-1]
        if obj in (True, False):
            return str(obj).lower()
        return str(obj)

    def parseJson(self, s: str):
        obj = json.loads(s)
        return AttributeDict(obj) if isinstance(obj, dict) else obj


class VelocityInput:
    """
    Simple class to mimic the behavior of variable '$input' in AWS API Gateway integration
    velocity templates.
    See: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
    """

    def __init__(self, body, params):
        self.parameters = params or {}
        self.value = body

    def _extract_json_path(self, path):
        if not self.value:
            return {}
        value = self.value if isinstance(self.value, dict) else json.loads(self.value)
        return extract_jsonpath(value, path)

    def path(self, path):
        return cast_to_vtl_json_object(self._extract_json_path(path))

    def json(self, path):
        path = path or "$"
        matching = self._extract_json_path(path)
        if isinstance(matching, (list, dict)):
            matching = json_safe(matching)
        return json.dumps(matching)

    @property
    def body(self):
        return self.value

    def params(self, name=None):
        if not name:
            return self.parameters
        for k in ["path", "querystring", "header"]:
            if val := self.parameters.get(k).get(name):
                return val
        return ""

    def __getattr__(self, name):
        return self.value.get(name)

    def __repr__(self):
        return "$input"


class ApiGatewayVtlTemplate(VtlTemplate):
    """Util class for rendering VTL templates with API Gateway specific extensions"""

    def prepare_namespace(self, variables, source: str = APIGW_SOURCE) -> dict[str, Any]:
        namespace = super().prepare_namespace(variables, source)
        input_var = variables.get("input") or {}
        variables = {
            "input": VelocityInput(input_var.get("body"), input_var.get("params")),
            "util": VelocityUtilApiGateway(),
        }
        namespace.update(variables)
        return namespace

    def render_request(
        self, template: str, variables: MappingTemplateVariables
    ) -> tuple[str, ContextVarsRequestOverride]:
        variables_copy: MappingTemplateVariables = copy.deepcopy(variables)
        variables_copy["context"]["requestOverride"] = ContextVarsRequestOverride(
            querystring={}, header={}, path={}
        )
        result = self.render_vtl(template=template.strip(), variables=variables_copy)
        return result, variables_copy["context"]["requestOverride"]

    def render_response(
        self, template: str, variables: MappingTemplateVariables
    ) -> tuple[str, ContextVarsResponseOverride]:
        variables_copy: MappingTemplateVariables = copy.deepcopy(variables)
        variables_copy["context"]["responseOverride"] = ContextVarsResponseOverride(
            header={}, status=0
        )
        result = self.render_vtl(template=template.strip(), variables=variables_copy)
        return result, variables_copy["context"]["responseOverride"]


# patches required to allow our custom class operations in VTL templates processed by airspeed
airspeed.operators.__additional_methods__[VTLMap] = airspeed.operators.__additional_methods__[dict]
airspeed.operators.__additional_methods__[VTLJsonDict] = airspeed.operators.__additional_methods__[
    dict
]
airspeed.operators.__additional_methods__[VTLJsonList] = airspeed.operators.__additional_methods__[
    list
]

1	# > In API Gateway, an API's method request or response can take a payload in a different format from the integration
2	# request or response.
3	#
4	# You can transform your data to:
5	# - Match the payload to an API-specified format.
6	# - Override an API's request and response parameters and status codes.
7	# - Return client selected response headers.
8	# - Associate path parameters, query string parameters, or header parameters in the method request of HTTP proxy
9	# or AWS service proxy. TODO: this is from the documentation. Can we use requestOverides for proxy integrations?
10	# - Select which data to send using integration with AWS services, such as Amazon DynamoDB or Lambda functions,
11	# or HTTP endpoints.
12	#
13	# You can use mapping templates to transform your data. A mapping template is a script expressed in Velocity Template
14	# Language (VTL) and applied to the payload using JSONPath .
15	#
16	# https://docs.aws.amazon.com/apigateway/latest/developerguide/models-mappings.html
17	import base64	1✔
18	import copy	1✔
19	import json	1✔
20	import logging	1✔
21	from typing import Any, TypedDict	1✔
22	from urllib.parse import quote_plus, unquote_plus	1✔
23
24	import airspeed	1✔
25	from airspeed.operators import dict_to_string	1✔
26
27	from localstack import config	1✔
28	from localstack.services.apigateway.next_gen.execute_api.variables import (	1✔
29	ContextVariables,
30	ContextVarsRequestOverride,
31	ContextVarsResponseOverride,
32	)
33	from localstack.utils.aws.templating import APIGW_SOURCE, VelocityUtil, VtlTemplate	1✔
34	from localstack.utils.json import extract_jsonpath, json_safe	1✔
35
36	LOG = logging.getLogger(__name__)	1✔
37
38
39	class MappingTemplateParams(TypedDict, total=False):	1✔
40	path: dict[str, str]	1✔
41	querystring: dict[str, str]	1✔
42	header: dict[str, str]	1✔
43
44
45	class MappingTemplateInput(TypedDict, total=False):	1✔
46	body: str	1✔
47	params: MappingTemplateParams	1✔
48
49
50	class MappingTemplateVariables(TypedDict, total=False):	1✔
51	context: ContextVariables	1✔
52	input: MappingTemplateInput	1✔
53	stageVariables: dict[str, str]	1✔
54
55
56	def cast_to_vtl_object(value):	1✔
NEW 57	if isinstance(value, dict):	×
NEW 58	return VTLMap(value)	×
NEW 59	if isinstance(value, list):	×
NEW 60	return [cast_to_vtl_object(item) for item in value]	×
NEW 61	return value	×
62
63
64	def cast_to_vtl_json_object(value: Any) -> Any:	1✔
65	if isinstance(value, dict):	1✔
66	return VTLJsonDict(value)	1✔
67	if isinstance(value, list):	1✔
68	return VTLJsonList(value)	1✔
69	return value	1✔
70
71
72	class VTLMap(dict):	1✔
73	"""Overrides __str__ of python dict (and all child dict) to return a Java like string representation"""
74
75	# TODO apply this class more generally through the template mappings
76
77	def __init__(self, args, *kwargs):	1✔
78	super().__init__(args, *kwargs)	1✔
79	self.update(args, *kwargs)	1✔
80
81	@staticmethod	1✔
82	def cast_factory(value: Any) -> Any:	1✔
NEW 83	return cast_to_vtl_object(value)	×
84
85	def update(self, args, *kwargs):	1✔
86	for k, v in self.items():	1✔
87	self[k] = self.cast_factory(v)	1✔
88
89	def __str__(self) -> str:	1✔
90	return dict_to_string(self)	1✔
91
92
93	class VTLJsonList(list):	1✔
94	"""Some VTL List behave differently when being represented as string and everything
95	inside will be represented as a json string
96
97	Example: $input.path('$').b // Where path is {"a": 1, "b": [{"c": 5}]}
98	Results: '[{"c":5}]' // Where everything inside the list is a valid json object
99	"""
100
101	def __init__(self, *args):	1✔
102	super(VTLJsonList, self).__init__(*args)	1✔
103	for idx, item in enumerate(self):	1✔
104	self[idx] = cast_to_vtl_json_object(item)	1✔
105
106	def __str__(self):	1✔
107	if isinstance(self, list):	1✔
108	return json.dumps(self, separators=(",", ":"))	1✔
109
110
111	class VTLJsonDict(VTLMap):	1✔
112	"""Some VTL Map behave differently when being represented as string and a list
113	encountered in the dictionary will be represented as a json string
114
115	Example: $input.path('$') // Where path is {"a": 1, "b": [{"c": 5}]}
116	Results: '{a=1, b=[{"c":5}]}' // Where everything inside the list is a valid json object
117	"""
118
119	@staticmethod	1✔
120	def cast_factory(value: Any) -> Any:	1✔
121	return cast_to_vtl_json_object(value)	1✔
122
123
124	class AttributeDict(dict):	1✔
125	"""
126	Wrapper returned by VelocityUtilApiGateway.parseJson to allow access to dict values as attributes (dot notation),
127	e.g.: $util.parseJson('$.foo').bar
128	"""
129
130	def __init__(self, args, *kwargs):	1✔
131	super(AttributeDict, self).__init__(args, *kwargs)	1✔
132	for key, value in self.items():	1✔
133	if isinstance(value, dict):	1✔
134	self[key] = AttributeDict(value)	1✔
135
136	def __getattr__(self, name):	1✔
137	if name in self:	1✔
138	return self[name]	1✔
139	raise AttributeError(f"'AttributeDict' object has no attribute '{name}'")	×
140
141	def __setattr__(self, name, value):	1✔
142	self[name] = value	×
143
144	def __delattr__(self, name):	1✔
145	if name in self:	×
146	del self[name]	×
147	else:
148	raise AttributeError(f"'AttributeDict' object has no attribute '{name}'")	×
149
150
151	class VelocityUtilApiGateway(VelocityUtil):	1✔
152	"""
153	Simple class to mimic the behavior of variable '$util' in AWS API Gateway integration
154	velocity templates.
155	See: https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
156	"""
157
158	def base64Encode(self, s):	1✔
159	if not isinstance(s, str):	1✔
160	s = json.dumps(s)	×
161	encoded_str = s.encode(config.DEFAULT_ENCODING)	1✔
162	encoded_b64_str = base64.b64encode(encoded_str)	1✔
163	return encoded_b64_str.decode(config.DEFAULT_ENCODING)	1✔
164
165	def base64Decode(self, s):	1✔
166	if not isinstance(s, str):	1✔
167	s = json.dumps(s)	×
168	return base64.b64decode(s)	1✔
169
170	def toJson(self, obj):	1✔
171	return obj and json.dumps(obj)	×
172
173	def urlEncode(self, s):	1✔
174	return quote_plus(s)	1✔
175
176	def urlDecode(self, s):	1✔
177	return unquote_plus(s)	1✔
178
179	def escapeJavaScript(self, obj: Any) -> str:	1✔
180	"""
181	Converts the given object to a string and escapes any regular single quotes (') into escaped ones (\').
182	JSON dumps will escape the single quotes.
183	https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
184	"""
185	if obj is None:	1✔
186	return "null"	1✔
187	if isinstance(obj, str):	1✔
188	# empty string escapes to empty object
189	if len(obj.strip()) == 0:	1✔
190	return "{}"	×
191	return json.dumps(obj)[1:-1]	1✔
192	if obj in (True, False):	1✔
193	return str(obj).lower()	1✔
194	return str(obj)	×
195
196	def parseJson(self, s: str):	1✔
197	obj = json.loads(s)	1✔
198	return AttributeDict(obj) if isinstance(obj, dict) else obj	1✔
199
200
201	class VelocityInput:	1✔
202	"""
203	Simple class to mimic the behavior of variable '$input' in AWS API Gateway integration
204	velocity templates.
205	See: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html
206	"""
207
208	def __init__(self, body, params):	1✔
209	self.parameters = params or {}	1✔
210	self.value = body	1✔
211
212	def _extract_json_path(self, path):	1✔
213	if not self.value:	1✔
214	return {}	1✔
215	value = self.value if isinstance(self.value, dict) else json.loads(self.value)	1✔
216	return extract_jsonpath(value, path)	1✔
217
218	def path(self, path):	1✔
219	return cast_to_vtl_json_object(self._extract_json_path(path))	1✔
220
221	def json(self, path):	1✔
222	path = path or "$"	1✔
223	matching = self._extract_json_path(path)	1✔
224	if isinstance(matching, (list, dict)):	1✔
225	matching = json_safe(matching)	1✔
226	return json.dumps(matching)	1✔
227
228	@property	1✔
229	def body(self):	1✔
230	return self.value	1✔
231
232	def params(self, name=None):	1✔
233	if not name:	1✔
234	return self.parameters	1✔
235	for k in ["path", "querystring", "header"]:	1✔
236	if val := self.parameters.get(k).get(name):	1✔
237	return val	1✔
238	return ""	1✔
239
240	def __getattr__(self, name):	1✔
241	return self.value.get(name)	×
242
243	def __repr__(self):
244	return "$input"
245
246
247	class ApiGatewayVtlTemplate(VtlTemplate):	1✔
248	"""Util class for rendering VTL templates with API Gateway specific extensions"""
249
250	def prepare_namespace(self, variables, source: str = APIGW_SOURCE) -> dict[str, Any]:	1✔
251	namespace = super().prepare_namespace(variables, source)	1✔
252	input_var = variables.get("input") or {}	1✔
253	variables = {	1✔
254	"input": VelocityInput(input_var.get("body"), input_var.get("params")),
255	"util": VelocityUtilApiGateway(),
256	}
257	namespace.update(variables)	1✔
258	return namespace	1✔
259
260	def render_request(	1✔
261	self, template: str, variables: MappingTemplateVariables
262	) -> tuple[str, ContextVarsRequestOverride]:
263	variables_copy: MappingTemplateVariables = copy.deepcopy(variables)	1✔
264	variables_copy["context"]["requestOverride"] = ContextVarsRequestOverride(	1✔
265	querystring={}, header={}, path={}
266	)
267	result = self.render_vtl(template=template.strip(), variables=variables_copy)	1✔
268	return result, variables_copy["context"]["requestOverride"]	1✔
269
270	def render_response(	1✔
271	self, template: str, variables: MappingTemplateVariables
272	) -> tuple[str, ContextVarsResponseOverride]:
273	variables_copy: MappingTemplateVariables = copy.deepcopy(variables)	1✔
274	variables_copy["context"]["responseOverride"] = ContextVarsResponseOverride(	1✔
275	header={}, status=0
276	)
277	result = self.render_vtl(template=template.strip(), variables=variables_copy)	1✔
278	return result, variables_copy["context"]["responseOverride"]	1✔
279
280
281	# patches required to allow our custom class operations in VTL templates processed by airspeed
282	airspeed.operators.__additional_methods__[VTLMap] = airspeed.operators.__additional_methods__[dict]	1✔
283	airspeed.operators.__additional_methods__[VTLJsonDict] = airspeed.operators.__additional_methods__[	1✔
284	dict
285	]
286	airspeed.operators.__additional_methods__[VTLJsonList] = airspeed.operators.__additional_methods__[	1✔
287	list
288	]

localstack / localstack / 65ebfa7b-38ab-49bb-b71e-44800576a11c

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