Merge pull request #186 from gyft/feat-case-insensitive-dict

feat(data_classes): case insensitive header lookup

Author: heitorlessa

aws_lambda_powertools/utilities/data_classes/common.py

@@ -48,7 +48,9 @@ def get_query_string_value(self, name: str, default_value: Optional[str] = None)
         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]:
+    def get_header_value(
+        self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False
+    ) -> Optional[str]:
         """Get header value by name
 
         Parameters
@@ -57,9 +59,14 @@ def get_header_value(self, name: str, default_value: Optional[str] = None) -> Op
             Header name
         default_value: str, optional
             Default value if no value was found by name
+        case_sensitive: bool
+            Whether to use a case sensitive look up
         Returns
         -------
         str, optional
             Header value
         """
-        return self.headers.get(name, default_value)
+        if case_sensitive:
+            return self.headers.get(name, default_value)
+
+        return next((value for key, value in self.headers.items() if name.lower() == key.lower()), default_value)

tests/functional/test_lambda_trigger_events.py

@@ -670,13 +670,40 @@ def test_base_proxy_event_get_header_value():
     value = event.get_header_value("test", default_value)
     assert value == set_value
 
+    # Verify that the default look is case insensitive
+    value = event.get_header_value("Test")
+    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_base_proxy_event_get_header_value_case_insensitive():
+    default_value = "default"
+    set_value = "value"
+
+    event = BaseProxyEvent({"headers": {}})
+
+    event._data["headers"] = {"Test": set_value}
+    value = event.get_header_value("test", case_sensitive=True)
+    assert value is None
+
+    value = event.get_header_value("test", default_value=default_value, case_sensitive=True)
+    assert value == default_value
+
+    value = event.get_header_value("Test", case_sensitive=True)
+    assert value == set_value
+
+    value = event.get_header_value("unknown", default_value, case_sensitive=True)
+    assert value == default_value
+
+    value = event.get_header_value("unknown", case_sensitive=True)
+    assert value is None
+
+
 def test_kinesis_stream_event():
     event = KinesisStreamEvent(load_event("kinesisStreamEvent.json"))