refactor(parser): add from __future__ import annotations (#4977)

* refactor(envelopes): add from __future__ import annotations

and update code according to ruff rules TCH, UP006, UP007, UP037 and
FA100.

* Fix e2e tests

---------

Co-authored-by: Leandro Damascena <[email protected]>

Author: ericbn

aws_lambda_powertools/utilities/parser/envelopes/apigw.py

@@ -1,24 +1,28 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventModel
 
-from ..models import APIGatewayProxyEventModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
 
 class ApiGatewayEnvelope(BaseEnvelope):
     """API Gateway envelope to extract data within body key"""
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
         """Parses data found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns

aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py

@@ -1,24 +1,28 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventV2Model
 
-from ..models import APIGatewayProxyEventV2Model
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
 
 class ApiGatewayV2Envelope(BaseEnvelope):
     """API Gateway V2 envelope to extract data within body key"""
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
         """Parses data found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns

aws_lambda_powertools/utilities/parser/envelopes/base.py

@@ -2,10 +2,12 @@
 
 import logging
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Optional, TypeVar, Union
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from aws_lambda_powertools.utilities.parser.functions import _retrieve_or_set_model_from_cache
-from aws_lambda_powertools.utilities.parser.types import T
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import T
 
 logger = logging.getLogger(__name__)
 
@@ -14,12 +16,12 @@ class BaseEnvelope(ABC):
     """ABC implementation for creating a supported Envelope"""
 
     @staticmethod
-    def _parse(data: Optional[Union[Dict[str, Any], Any]], model: type[T]) -> Union[T, None]:
+    def _parse(data: dict[str, Any] | Any | None, model: type[T]) -> T | None:
         """Parses envelope data against model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Data to be parsed and validated
         model : type[T]
             Data model to parse and validate data against
@@ -43,7 +45,7 @@ def _parse(data: Optional[Union[Dict[str, Any], Any]], model: type[T]) -> Union[
         return adapter.validate_python(data)
 
     @abstractmethod
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: type[T]):
+    def parse(self, data: dict[str, Any] | Any | None, model: type[T]):
         """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

@@ -1,29 +1,33 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import BedrockAgentEventModel
 
-from ..models import BedrockAgentEventModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
 
 class BedrockAgentEnvelope(BaseEnvelope):
     """Bedrock Agent envelope to extract data within input_text key"""
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
         """Parses data found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        Optional[Model]
+        Model | None
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Bedrock Agent model {BedrockAgentEventModel}")

aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py

@@ -1,15 +1,19 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import CloudWatchLogsModel
 
-from ..models import CloudWatchLogsModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
 
 class CloudWatchLogsEnvelope(BaseEnvelope):
-    """CloudWatch Envelope to extract a List of log records.
+    """CloudWatch Envelope to extract a list of log records.
 
     The record's body parameter is a string (after being base64 decoded and gzipped),
     though it can also be a JSON encoded string.
@@ -18,19 +22,19 @@ class CloudWatchLogsEnvelope(BaseEnvelope):
     Note: The record will be parsed the same way so if model is str
     """
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
         """Parses records found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        List
+        list
             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

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamModel
 
-from ..models import DynamoDBStreamModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
@@ -15,19 +19,19 @@ class DynamoDBStreamEnvelope(BaseEnvelope):
     length of the list is the record's amount in the original event.
     """
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Dict[str, Optional[Model]]]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[dict[str, Model | None]]:
         """Parses DynamoDB Stream records found in either NewImage and OldImage with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        List
+        list
             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

@@ -1,24 +1,28 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, Optional, Type, Union
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
 
-from ..models import EventBridgeModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
 
 class EventBridgeEnvelope(BaseEnvelope):
     """EventBridge envelope to extract data within detail key"""
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None:
         """Parses data found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns

aws_lambda_powertools/utilities/parser/envelopes/kafka.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, List, Optional, Type, Union, cast
+from typing import TYPE_CHECKING, Any, cast
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import KafkaMskEventModel, KafkaSelfManagedEventModel
 
-from ..models import KafkaMskEventModel, KafkaSelfManagedEventModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
@@ -17,23 +21,23 @@ class KafkaEnvelope(BaseEnvelope):
     all items in the list will be parsed as str and npt as JSON (and vice versa)
     """
 
-    def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
         """Parses data found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        List
+        list
             List of records parsed with model provided
         """
         event_source = cast(dict, data).get("eventSource")
-        model_parse_event: Union[Type[KafkaMskEventModel], Type[KafkaSelfManagedEventModel]] = (
+        model_parse_event: type[KafkaMskEventModel | KafkaSelfManagedEventModel] = (
             KafkaMskEventModel if event_source == "aws:kafka" else KafkaSelfManagedEventModel
         )
 

aws_lambda_powertools/utilities/parser/envelopes/kinesis.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, List, Optional, Type, Union, cast
+from typing import TYPE_CHECKING, Any, cast
+
+from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope
+from aws_lambda_powertools.utilities.parser.models import KinesisDataStreamModel
 
-from ..models import KinesisDataStreamModel
-from ..types import Model
-from .base import BaseEnvelope
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import Model
 
 logger = logging.getLogger(__name__)
 
@@ -19,19 +23,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: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]:
+    def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]:
         """Parses records found with model provided
 
         Parameters
         ----------
-        data : Dict
+        data : dict
             Lambda event to be parsed
-        model : Type[Model]
+        model : type[Model]
             Data model provided to parse after extracting data using envelope
 
         Returns
         -------
-        List
+        list
             List of records parsed with model provided
         """
         logger.debug(f"Parsing incoming data with Kinesis model {KinesisDataStreamModel}")