feat(trigger): API gateway helper methods

Author: Michael Brewer

aws_lambda_powertools/utilities/trigger/api_gateway_proxy_event.py

@@ -257,6 +257,39 @@ def body(self) -> Optional[str]:
     def is_base64_encoded(self) -> bool:
         return self["isBase64Encoded"]
 
+    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        """Get query string value by name
+
+        Parameters
+        ----------
+        name: str
+            Query string parameter name
+        default_value: str, optional
+            Default value if no value was found by name
+        Returns
+        -------
+        str, optional
+            Query string parameter value
+        """
+        params = self.query_string_parameters
+        return default_value if params is None else params.get(name, default_value)
+
+    def get_header_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return self.headers.get(name, default_value)
+
 
 class RequestContextV2Http(DictWrapper):
     @property
@@ -412,3 +445,36 @@ def is_base64_encoded(self) -> bool:
     @property
     def stage_variables(self) -> Optional[Dict[str, str]]:
         return self.get("stageVariables")
+
+    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        """Get query string value by name
+
+        Parameters
+        ----------
+        name: str
+            Query string parameter name
+        default_value: str, optional
+            Default value if no value was found by name
+        Returns
+        -------
+        str, optional
+            Query string parameter value
+        """
+        params = self.query_string_parameters
+        return default_value if params is None else params.get(name, default_value)
+
+    def get_header_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return self.headers.get(name, default_value)

tests/functional/test_lambda_trigger_events.py

@@ -527,3 +527,79 @@ def test_api_gateway_proxy_v2_event():
     assert event.path_parameters == event["pathParameters"]
     assert event.is_base64_encoded == event["isBase64Encoded"]
     assert event.stage_variables == event["stageVariables"]
+
+
+def test_api_gateway_proxy_get_query_string_value():
+    default_value = "default"
+    set_value = "value"
+
+    event = APIGatewayProxyEvent({})
+    value = event.get_query_string_value("test", default_value)
+    assert value == default_value
+
+    event["queryStringParameters"] = {"test": set_value}
+    value = event.get_query_string_value("test", default_value)
+    assert value == set_value
+
+    value = event.get_query_string_value("unknown", default_value)
+    assert value == default_value
+
+    value = event.get_query_string_value("unknown")
+    assert value is None
+
+
+def test_api_gateway_proxy_v2_get_query_string_value():
+    default_value = "default"
+    set_value = "value"
+
+    event = APIGatewayProxyEventV2({})
+    value = event.get_query_string_value("test", default_value)
+    assert value == default_value
+
+    event["queryStringParameters"] = {"test": set_value}
+    value = event.get_query_string_value("test", default_value)
+    assert value == set_value
+
+    value = event.get_query_string_value("unknown", default_value)
+    assert value == default_value
+
+    value = event.get_query_string_value("unknown")
+    assert value is None
+
+
+def test_api_gateway_proxy_get_header_value():
+    default_value = "default"
+    set_value = "value"
+
+    event = APIGatewayProxyEvent({"headers": {}})
+    value = event.get_header_value("test", default_value)
+    assert value == default_value
+
+    event["headers"] = {"test": set_value}
+    value = event.get_header_value("test", default_value)
+    assert value == set_value
+
+    value = event.get_header_value("unknown", default_value)
+    assert value == default_value
+
+    value = event.get_header_value("unknown")
+    assert value is None
+
+
+def test_api_gateway_proxy_v2_get_header_value():
+    default_value = "default"
+    set_value = "value"
+
+    event = APIGatewayProxyEventV2({"headers": {}})
+    value = event.get_header_value("test", default_value)
+    assert value == default_value
+
+    event["headers"] = {"test": set_value}
+    value = event.get_header_value("test", default_value)
+    assert value == set_value
+
+    value = event.get_header_value("unknown", default_value)
+    assert value == default_value
+
+    value = event.get_header_value("unknown")
+    assert value is None