Add custom CaseInsensitiveDict

This is hopefully a simpler implementations that the requests' package
one, but still had to be minimally complex to be complete.

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -1,6 +1,4 @@
-from typing import Any, Dict, List, MutableMapping
-
-from requests.structures import CaseInsensitiveDict
+from typing import Any, Dict, List
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
@@ -9,6 +7,7 @@
 )
 from aws_lambda_powertools.utilities.data_classes.common import (
     BaseProxyEvent,
+    CaseInsensitiveDict,
     DictWrapper,
 )
 
@@ -42,11 +41,11 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return self.multi_value_query_string_parameters or super().resolved_query_string_parameters
 
     @property
-    def multi_value_headers(self) -> MutableMapping[str, List[str]]:
+    def multi_value_headers(self) -> Dict[str, List[str]]:
         return CaseInsensitiveDict(self.get("multiValueHeaders"))
 
     @property
-    def resolved_headers_field(self) -> MutableMapping[str, Any]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         return self.multi_value_headers or self.headers
 
     def header_serializer(self) -> BaseHeadersSerializer:

aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py

@@ -1,12 +1,11 @@
 import enum
 import re
-from typing import Any, Dict, List, MutableMapping, Optional
-
-from requests.structures import CaseInsensitiveDict
+from typing import Any, Dict, List, Optional
 
 from aws_lambda_powertools.utilities.data_classes.common import (
     BaseRequestContext,
     BaseRequestContextV2,
+    CaseInsensitiveDict,
     DictWrapper,
 )
 
@@ -142,7 +141,7 @@ def http_method(self) -> str:
         return self["httpMethod"]
 
     @property
-    def headers(self) -> MutableMapping[str, str]:
+    def headers(self) -> Dict[str, str]:
         return CaseInsensitiveDict(self["headers"])
 
     @property
@@ -223,7 +222,7 @@ def cookies(self) -> List[str]:
         return self["cookies"]
 
     @property
-    def headers(self) -> MutableMapping[str, str]:
+    def headers(self) -> Dict[str, str]:
         """Http headers"""
         return CaseInsensitiveDict(self["headers"])
 

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -1,7 +1,5 @@
 from functools import cached_property
-from typing import Any, Dict, List, MutableMapping, Optional
-
-from requests.structures import CaseInsensitiveDict
+from typing import Any, Dict, List, Optional
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
@@ -12,6 +10,7 @@
     BaseProxyEvent,
     BaseRequestContext,
     BaseRequestContextV2,
+    CaseInsensitiveDict,
     DictWrapper,
 )
 
@@ -115,7 +114,7 @@ def resource(self) -> str:
         return self["resource"]
 
     @property
-    def multi_value_headers(self) -> MutableMapping[str, List[str]]:
+    def multi_value_headers(self) -> Dict[str, List[str]]:
         return CaseInsensitiveDict(self.get("multiValueHeaders"))
 
     @property
@@ -130,7 +129,7 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return super().resolved_query_string_parameters
 
     @property
-    def resolved_headers_field(self) -> MutableMapping[str, Any]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         return self.multi_value_headers or self.headers
 
     @property
@@ -316,5 +315,5 @@ def header_serializer(self):
         return HttpApiHeadersSerializer()
 
     @cached_property
-    def resolved_headers_field(self) -> MutableMapping[str, Any]:
-        return CaseInsensitiveDict({k: v.split(",") if "," in v else v for k, v in self.headers.items()})
+    def resolved_headers_field(self) -> Dict[str, Any]:
+        return CaseInsensitiveDict((k, v.split(",") if "," in v else v) for k, v in self.headers.items())

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -1,8 +1,6 @@
-from typing import Any, Dict, List, MutableMapping, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
-from requests.structures import CaseInsensitiveDict
-
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict, DictWrapper
 
 
 def get_identity_object(identity: Optional[dict]) -> Any:
@@ -188,7 +186,7 @@ def source(self) -> Dict[str, Any]:
         return self.get("source") or {}
 
     @property
-    def request_headers(self) -> MutableMapping[str, str]:
+    def request_headers(self) -> Dict[str, str]:
         """Request headers"""
         return CaseInsensitiveDict(self["request"]["headers"])
 

aws_lambda_powertools/utilities/data_classes/common.py

@@ -1,13 +1,55 @@
 import base64
 import json
 from functools import cached_property
-from typing import Any, Callable, Dict, Iterator, List, Mapping, MutableMapping, Optional
-
-from requests.structures import CaseInsensitiveDict
+from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional
 
 from aws_lambda_powertools.shared.headers_serializer import BaseHeadersSerializer
 
 
+class CaseInsensitiveDict(dict):
+    """Case insensitive dict implementation. Assumes string keys only."""
+
+    def __init__(self, data=None, **kwargs):
+        super().__init__()
+        self.update(data, **kwargs)
+
+    def get(self, k, default=None):
+        return super().get(k.lower(), default)
+
+    def pop(self, k):
+        return super().pop(k.lower())
+
+    def setdefault(self, k, default=None):
+        return super().setdefault(k.lower(), default)
+
+    def update(self, data=None, **kwargs):
+        if data is not None:
+            if isinstance(data, Mapping):
+                super().update((k.lower(), v) for k, v in data.items())
+            else:
+                super().update((k.lower(), v) for k, v in data)
+        super().update((k.lower(), v) for k, v in kwargs)
+
+    def __contains__(self, k):
+        return super().__contains__(k.lower())
+
+    def __delitem__(self, k):
+        super().__delitem__(k.lower())
+
+    def __eq__(self, other):
+        if not isinstance(other, Mapping):
+            return False
+        if not isinstance(other, CaseInsensitiveDict):
+            other = CaseInsensitiveDict(other)
+        return super().__eq__(other)
+
+    def __getitem__(self, k):
+        return super().__getitem__(k.lower())
+
+    def __setitem__(self, k, v):
+        super().__setitem__(k.lower(), v)
+
+
 class DictWrapper(Mapping):
     """Provides a single read only access to a wrapper dict"""
 
@@ -93,7 +135,7 @@ def raw_event(self) -> Dict[str, Any]:
 
 class BaseProxyEvent(DictWrapper):
     @property
-    def headers(self) -> MutableMapping[str, str]:
+    def headers(self) -> Dict[str, str]:
         return CaseInsensitiveDict(self.get("headers"))
 
     @property
@@ -116,7 +158,7 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return {k: v.split(",") for k, v in self.query_string_parameters.items()}
 
     @property
-    def resolved_headers_field(self) -> MutableMapping[str, str]:
+    def resolved_headers_field(self) -> Dict[str, str]:
         """
         This property determines the appropriate header to be used
         as a trusted source for validating OpenAPI.

aws_lambda_powertools/utilities/data_classes/kafka_event.py

@@ -1,10 +1,8 @@
 import base64
 from functools import cached_property
-from typing import Any, Dict, Iterator, List, MutableMapping, Optional
+from typing import Any, Dict, Iterator, List, Optional
 
-from requests.structures import CaseInsensitiveDict
-
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict, DictWrapper
 
 
 class KafkaEventRecord(DictWrapper):
@@ -64,9 +62,9 @@ def headers(self) -> List[Dict[str, List[int]]]:
         return self["headers"]
 
     @cached_property
-    def decoded_headers(self) -> MutableMapping[str, bytes]:
+    def decoded_headers(self) -> Dict[str, bytes]:
         """Decodes the headers as a single dictionary."""
-        return CaseInsensitiveDict({k: bytes(v) for chunk in self.headers for k, v in chunk.items()})
+        return CaseInsensitiveDict((k, bytes(v)) for chunk in self.headers for k, v in chunk.items())
 
 
 class KafkaEvent(DictWrapper):

aws_lambda_powertools/utilities/data_classes/s3_object_event.py

@@ -1,8 +1,6 @@
-from typing import MutableMapping, Optional
+from typing import Dict, Optional
 
-from requests.structures import CaseInsensitiveDict
-
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict, DictWrapper
 
 
 class S3ObjectContext(DictWrapper):
@@ -64,7 +62,7 @@ def url(self) -> str:
         return self["url"]
 
     @property
-    def headers(self) -> MutableMapping[str, str]:
+    def headers(self) -> Dict[str, str]:
         """A map of string to strings containing the HTTP headers and their values from the original call,
         excluding any authorization-related headers.
 

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -1,13 +1,15 @@
 from functools import cached_property
-from typing import Any, Dict, MutableMapping, Optional
-
-from requests.structures import CaseInsensitiveDict
+from typing import Any, Dict, Optional
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
     HttpApiHeadersSerializer,
 )
-from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
+from aws_lambda_powertools.utilities.data_classes.common import (
+    BaseProxyEvent,
+    CaseInsensitiveDict,
+    DictWrapper,
+)
 from aws_lambda_powertools.utilities.data_classes.shared_functions import base64_decode
 
 
@@ -23,7 +25,7 @@ def json_body(self) -> Any:
         return self._json_deserializer(self.decoded_body)
 
     @property
-    def headers(self) -> MutableMapping[str, str]:
+    def headers(self) -> Dict[str, str]:
         """The VPC Lattice event headers."""
         return CaseInsensitiveDict(self["headers"])
 
@@ -73,8 +75,8 @@ def query_string_parameters(self) -> Dict[str, str]:
         return self["query_string_parameters"]
 
     @cached_property
-    def resolved_headers_field(self) -> MutableMapping[str, Any]:
-        return CaseInsensitiveDict({k: v.split(",") if "," in v else v for k, v in self.headers.items()})
+    def resolved_headers_field(self) -> Dict[str, Any]:
+        return CaseInsensitiveDict((k, v.split(",") if "," in v else v) for k, v in self.headers.items())
 
 
 class vpcLatticeEventV2Identity(DictWrapper):