fix(typing): resolved_headers_field is not Optional (#4148)

Co-authored-by: Heitor Lessa <lessa@amazon.com>

Author: tonnico

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1646,13 +1646,13 @@ def _determine_openapi_version(openapi_version):
         from aws_lambda_powertools.event_handler.openapi.pydantic_loader import PYDANTIC_V2
 
         # Pydantic V2 has no support for OpenAPI schema 3.0
-        if PYDANTIC_V2 and not openapi_version.startswith("3.1"):
+        if PYDANTIC_V2 and not openapi_version.startswith("3.1"):  # pragma: no cover
             warnings.warn(
                 "You are using Pydantic v2, which is incompatible with OpenAPI schema 3.0. Forcing OpenAPI 3.1",
                 stacklevel=2,
             )
             openapi_version = "3.1.0"
-        elif not PYDANTIC_V2 and not openapi_version.startswith("3.0"):
+        elif not PYDANTIC_V2 and not openapi_version.startswith("3.0"):  # pragma: no cover
             warnings.warn(
                 "You are using Pydantic v1, which is incompatible with OpenAPI schema 3.1. Forcing OpenAPI 3.0",
                 stacklevel=2,
@@ -2192,7 +2192,7 @@ def not_found(self, func: Optional[Callable] = None):
 
     def exception_handler(self, exc_class: Union[Type[Exception], List[Type[Exception]]]):
         def register_exception_handler(func: Callable):
-            if isinstance(exc_class, list):
+            if isinstance(exc_class, list):  # pragma: no cover
                 for exp in exc_class:
                     self._exception_handlers[exp] = func
             else:

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -410,7 +410,7 @@ def _normalize_multi_query_string_with_param(
     return resolved_query_string
 
 
-def _normalize_multi_header_values_with_param(headers: Optional[Dict[str, str]], params: Sequence[ModelField]):
+def _normalize_multi_header_values_with_param(headers: Dict[str, Any], params: Sequence[ModelField]):
     """
     Extract and normalize resolved_headers_field
 

aws_lambda_powertools/event_handler/openapi/compat.py

@@ -40,7 +40,7 @@
 
 RequestErrorModel: Type[BaseModel] = create_model("Request")
 
-if PYDANTIC_V2:
+if PYDANTIC_V2:  # pragma: no cover # false positive; dropping in v3
     from pydantic import TypeAdapter, ValidationError
     from pydantic._internal._typing_extra import eval_type_lenient
     from pydantic.fields import FieldInfo

aws_lambda_powertools/event_handler/openapi/pydantic_loader.py

@@ -3,4 +3,4 @@
 
     PYDANTIC_V2 = PYDANTIC_VERSION.startswith("2.")
 except ImportError:
-    PYDANTIC_V2 = False
+    PYDANTIC_V2 = False  # pragma: no cover  # false positive; dropping in v3

aws_lambda_powertools/utilities/batch/types.py

@@ -7,7 +7,7 @@
 
 # For IntelliSense and Mypy to work, we need to account for possible SQS subclasses
 # We need them as subclasses as we must access their message ID or sequence number metadata via dot notation
-if has_pydantic:
+if has_pydantic:  # pragma: no cover
     from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamRecordModel, SqsRecordModel
     from aws_lambda_powertools.utilities.parser.models import (
         KinesisDataStreamRecord as KinesisDataStreamRecordModel,
@@ -17,7 +17,7 @@
         Union[Type[SqsRecordModel], Type[DynamoDBStreamRecordModel], Type[KinesisDataStreamRecordModel]]
     ]
     BatchSqsTypeModel = Optional[Type[SqsRecordModel]]
-else:
+else:  # pragma: no cover
     BatchTypeModels = "BatchTypeModels"  # type: ignore
     BatchSqsTypeModel = "BatchSqsTypeModel"  # type: ignore
 

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -43,7 +43,7 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return super().resolved_query_string_parameters
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         headers: Dict[str, Any] = {}
 
         if self.multi_value_headers:

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -126,7 +126,7 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return super().resolved_query_string_parameters
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         headers: Dict[str, Any] = {}
 
         if self.multi_value_headers:
@@ -319,7 +319,7 @@ def header_serializer(self):
         return HttpApiHeadersSerializer()
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         if self.headers is not None:
             headers = {key.lower(): value.split(",") if "," in value else value for key, value in self.headers.items()}
             return headers

aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py

@@ -111,7 +111,7 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
         return {x["name"]: x["value"] for x in self["parameters"]} if self.get("parameters") else None
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         return {}
 
     @cached_property

aws_lambda_powertools/utilities/data_classes/common.py

@@ -124,7 +124,7 @@ def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
         return {}
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         """
         This property determines the appropriate header to be used
         as a trusted source for validating OpenAPI.

aws_lambda_powertools/utilities/data_classes/vpc_lattice.py

@@ -145,7 +145,7 @@ def query_string_parameters(self) -> Dict[str, str]:
         return self["query_string_parameters"]
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, Any]]:
+    def resolved_headers_field(self) -> Dict[str, Any]:
         if self.headers is not None:
             headers = {key.lower(): value.split(",") if "," in value else value for key, value in self.headers.items()}
             return headers
@@ -272,7 +272,7 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]:
             return None
 
     @property
-    def resolved_headers_field(self) -> Optional[Dict[str, str]]:
+    def resolved_headers_field(self) -> Dict[str, str]:
         if self.headers is not None:
             return {key.lower(): value for key, value in self.headers.items()}