fix(parser): type hints should reflect primitive types support (#8175)

* fix: parser type hints should reflect primitive types

As per #4502, primitive types are supported by the `parser` utility. However, the type hints utilized in the functions and envelopes there do not reflect that at the moment, making type checkers lack when inferring the return types, for example. This aims to improve this situation by adjusting the type hints there.

* fix: small changes

---------

Co-authored-by: Leandro Damascena <lcdama@amazon.pt>

Author: catarinacps

aws_lambda_powertools/utilities/parser/envelopes/apigw.py

@@ -7,27 +7,27 @@
 from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
 
 class ApiGatewayEnvelope(BaseEnvelope):
     """API Gateway envelope to extract data within body key"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Any
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Api Gateway model {APIGatewayProxyEventModel}")

aws_lambda_powertools/utilities/parser/envelopes/apigw_websocket.py

@@ -7,7 +7,7 @@
 from aws_lambda_powertools.utilities.parser.models import APIGatewayWebSocketMessageEventModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -16,19 +16,19 @@ class ApiGatewayWebSocketEnvelope(BaseEnvelope):
     """API Gateway WebSockets envelope to extract data within body key of messages routes
     (not disconnect or connect)"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Any
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(

aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py

@@ -7,27 +7,27 @@
 from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventV2Model
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
 
 class ApiGatewayV2Envelope(BaseEnvelope):
     """API Gateway V2 envelope to extract data within body key"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Any
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Api Gateway model V2 {APIGatewayProxyEventV2Model}")

aws_lambda_powertools/utilities/parser/envelopes/base.py

@@ -44,7 +44,11 @@ def _parse(data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         return _parse_and_validate_event(data=data, adapter=adapter)
 
     @abstractmethod
-    def parse(self, data: dict[str, Any] | Any | None, model: type[T]):
+    def parse(
+        self,
+        data: dict[str, Any] | Any | None,
+        model: type[T],
+    ) -> T | list[T | None] | list[dict[str, T | None]] | None:
         """Implementation to parse data against envelope model, then against the data model
 
         NOTE: Call `_parse` method to fully parse data with model provided.

aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py

@@ -7,27 +7,27 @@
 from aws_lambda_powertools.utilities.parser.models import BedrockAgentEventModel, BedrockAgentFunctionEventModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
 
 class BedrockAgentEnvelope(BaseEnvelope):
     """Bedrock Agent envelope to extract data within input_text key"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Model | None
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Bedrock Agent model {BedrockAgentEventModel}")
@@ -39,19 +39,19 @@ def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model
 class BedrockAgentFunctionEnvelope(BaseEnvelope):
     """Bedrock Agent Function envelope to extract data within input_text key"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Model | None
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Bedrock Agent Function model {BedrockAgentFunctionEventModel}")

aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py

@@ -7,7 +7,7 @@
 from aws_lambda_powertools.utilities.parser.models import CloudWatchLogsModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -22,19 +22,19 @@ class CloudWatchLogsEnvelope(BaseEnvelope):
     Note: The record will be parsed the same way so if model is str
     """
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]:
         """Parses records found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        list
+        list[T | None]
             List of records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with SNS model {CloudWatchLogsModel}")

aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py

@@ -7,7 +7,7 @@
 from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -19,19 +19,19 @@ class DynamoDBStreamEnvelope(BaseEnvelope):
     length of the list is the record's amount in the original event.
     """
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[dict[str, Model | None]]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[dict[str, T | None]]:
         """Parses DynamoDB Stream records found in either NewImage and OldImage with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        list
+        list[T | None]
             List of dictionaries with NewImage and OldImage records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with DynamoDB Stream model {DynamoDBStreamModel}")

aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py

@@ -7,27 +7,27 @@
 from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
 
 class EventBridgeEnvelope(BaseEnvelope):
     """EventBridge envelope to extract data within detail key"""
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Any
+        T | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with EventBridge model {EventBridgeModel}")

aws_lambda_powertools/utilities/parser/envelopes/kafka.py

@@ -7,7 +7,7 @@
 from aws_lambda_powertools.utilities.parser.models import KafkaMskEventModel, KafkaSelfManagedEventModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -21,19 +21,19 @@ class KafkaEnvelope(BaseEnvelope):
     all items in the list will be parsed as str and npt as JSON (and vice versa)
     """
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]:
         """Parses data found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        list
+        list[T | None]
             List of records parsed with model provided
         """
         event_source = cast(dict, data).get("eventSource")

aws_lambda_powertools/utilities/parser/envelopes/kinesis.py

@@ -8,7 +8,7 @@
 from aws_lambda_powertools.utilities.parser.models import KinesisDataStreamModel
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.parser.types import Model
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -24,19 +24,19 @@ class KinesisDataStreamEnvelope(BaseEnvelope):
     all items in the list will be parsed as str and not as JSON (and vice versa)
     """
 
-    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]:
         """Parses records found with model provided
 
         Parameters
         ----------
         data : dict
             Lambda event to be parsed
-        model : type[Model]
+        model : type[T]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        list
+        list[T | None]
             List of records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with Kinesis model {KinesisDataStreamModel}")