refactor(event_source): use standard collections for types (#6479)

Using generics types

Co-authored-by: Ana Falcão <[email protected]>

Author: leandrodamascena

aws_lambda_powertools/utilities/data_classes/active_mq_event.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 from functools import cached_property
-from typing import Any, Iterator
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 from aws_lambda_powertools.utilities.data_classes.shared_functions import base64_decode
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class ActiveMQMessage(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/common.py

@@ -9,14 +9,17 @@
 import base64
 import json
 import warnings
+from collections.abc import Mapping
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, Callable, Iterator, Mapping, overload
+from typing import TYPE_CHECKING, Any, overload
 
 from typing_extensions import deprecated
 
 from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning
 
 if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator
+
     from aws_lambda_powertools.shared.headers_serializer import BaseHeadersSerializer
 
 from aws_lambda_powertools.utilities.data_classes.shared_functions import (

aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py

@@ -2,11 +2,14 @@
 
 from enum import Enum
 from functools import cached_property
-from typing import Any, Iterator
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.shared.dynamodb_deserializer import TypeDeserializer
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class StreamViewType(Enum):
     """The type of data from the modified DynamoDB item that was captured in this stream record"""

aws_lambda_powertools/utilities/data_classes/event_source.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
     from aws_lambda_powertools.utilities.typing import LambdaContext
 

aws_lambda_powertools/utilities/data_classes/kafka_event.py

@@ -2,10 +2,13 @@
 
 import base64
 from functools import cached_property
-from typing import Any, Iterator
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.utilities.data_classes.common import CaseInsensitiveDict, DictWrapper
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class KafkaEventRecord(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/kinesis_firehose_event.py

@@ -5,11 +5,13 @@
 import warnings
 from dataclasses import dataclass, field
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, Callable, ClassVar, Iterator
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
 if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator
+
     from typing_extensions import Literal
 
 

aws_lambda_powertools/utilities/data_classes/kinesis_stream_event.py

@@ -3,13 +3,16 @@
 import base64
 import json
 import zlib
-from typing import Iterator
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import (
     CloudWatchLogsDecodedData,
 )
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class KinesisStreamRecordPayload(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/s3_batch_operation_event.py

@@ -2,7 +2,7 @@
 
 import warnings
 from dataclasses import dataclass, field
-from typing import Any, Iterator, Literal
+from typing import TYPE_CHECKING, Any, Literal
 from urllib.parse import unquote_plus
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
@@ -11,6 +11,9 @@
 VALID_RESULT_CODES: tuple[str, str, str] = ("Succeeded", "TemporaryFailure", "PermanentFailure")
 RESULT_CODE_TYPE = Literal["Succeeded", "TemporaryFailure", "PermanentFailure"]
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 @dataclass(repr=False, order=False)
 class S3BatchOperationResponseRecord:

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
-from typing import Iterator
+from typing import TYPE_CHECKING
 from urllib.parse import unquote_plus
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 from aws_lambda_powertools.utilities.data_classes.event_bridge_event import (
     EventBridgeEvent,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class S3Identity(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/ses_event.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
-from typing import Iterator
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class SESMailHeader(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/sns_event.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
-from typing import Iterator
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class SNSMessageAttribute(DictWrapper):
     @property

aws_lambda_powertools/utilities/data_classes/sqs_event.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 from functools import cached_property
-from typing import Any, Dict, ItemsView, Iterator, TypeVar
+from typing import TYPE_CHECKING, Any, ItemsView, Iterator, TypeVar
 
 from aws_lambda_powertools.utilities.data_classes import S3Event
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 from aws_lambda_powertools.utilities.data_classes.sns_event import SNSMessage
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
 
 class SQSRecordAttributes(DictWrapper):
     @property
@@ -86,7 +89,7 @@ def data_type(self) -> str:
         return self["dataType"]
 
 
-class SQSMessageAttributes(Dict[str, SQSMessageAttribute]):
+class SQSMessageAttributes(dict[str, SQSMessageAttribute]):
     def __getitem__(self, key: str) -> SQSMessageAttribute | None:  # type: ignore
         item = super().get(key)
         return None if item is None else SQSMessageAttribute(item)  # type: ignore

tests/unit/data_classes/_boto3/test_code_pipeline_job_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 import zipfile
 from io import StringIO

tests/unit/data_classes/required_dependencies/test_active_mq_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes.active_mq_event import (

tests/unit/data_classes/required_dependencies/test_alb_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import ALBEvent
 from tests.functional.utils import load_event
 

tests/unit/data_classes/required_dependencies/test_api_gateway_authorizer.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (

tests/unit/data_classes/required_dependencies/test_api_gateway_authorizer_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (

tests/unit/data_classes/required_dependencies/test_api_gateway_authorizer_websocket_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (

tests/unit/data_classes/required_dependencies/test_api_gateway_proxy_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import (
     APIGatewayProxyEvent,
     APIGatewayProxyEventV2,

tests/unit/data_classes/required_dependencies/test_api_gateway_websocket_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from aws_lambda_powertools.utilities.data_classes import APIGatewayWebSocketEvent

tests/unit/data_classes/required_dependencies/test_appsync_authorizer_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes.appsync_authorizer_event import (
     AppSyncAuthorizerEvent,
     AppSyncAuthorizerResponse,

tests/unit/data_classes/required_dependencies/test_appsync_resolver_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent

tests/unit/data_classes/required_dependencies/test_aws_config_rule_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from aws_lambda_powertools.utilities.data_classes import AWSConfigRuleEvent

tests/unit/data_classes/required_dependencies/test_bedrock_agent_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent
 from tests.functional.utils import load_event
 

tests/unit/data_classes/required_dependencies/test_cloud_watch_alarm_event.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 import json
-from typing import Dict, List
 
 from aws_lambda_powertools.utilities.data_classes import CloudWatchAlarmEvent
 from tests.functional.utils import load_event
@@ -36,7 +37,7 @@ def test_cloud_watch_alarm_event_single_metric():
     assert parsed_event.alarm_data.configuration.alarm_actions_suppressor_extension_period is None
     assert parsed_event.alarm_data.configuration.alarm_actions_suppressor_wait_period is None
 
-    assert isinstance(parsed_event.alarm_data.configuration.metrics, List)
+    assert isinstance(parsed_event.alarm_data.configuration.metrics, list)
     # metric position 0
     metric_0 = parsed_event.alarm_data.configuration.metrics[0]
     raw_metric_0 = raw_event["alarmData"]["configuration"]["metrics"][0]
@@ -53,7 +54,7 @@ def test_cloud_watch_alarm_event_single_metric():
     assert metric_1.metric_stat.stat == raw_metric_1["metricStat"]["stat"]
     assert metric_1.metric_stat.period == raw_metric_1["metricStat"]["period"]
     assert metric_1.metric_stat.unit is None
-    assert isinstance(metric_1.metric_stat.metric, Dict)
+    assert isinstance(metric_1.metric_stat.metric, dict)
 
 
 def test_cloud_watch_alarm_event_composite_metric():
@@ -102,4 +103,4 @@ def test_cloud_watch_alarm_event_composite_metric():
         parsed_event.alarm_data.configuration.alarm_actions_suppressor
         == raw_event["alarmData"]["configuration"]["actionsSuppressor"]
     )
-    assert isinstance(parsed_event.alarm_data.configuration.metrics, List)
+    assert isinstance(parsed_event.alarm_data.configuration.metrics, list)

tests/unit/data_classes/required_dependencies/test_cloud_watch_custom_widget_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import (
     CloudWatchDashboardCustomWidgetEvent,
 )

tests/unit/data_classes/required_dependencies/test_cloud_watch_logs_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import CloudWatchLogsEvent
 from tests.functional.utils import load_event
 

tests/unit/data_classes/required_dependencies/test_cloudformation_custom_resource_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes import (

tests/unit/data_classes/required_dependencies/test_code_deploy_lifecycle_hook_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes import (

tests/unit/data_classes/required_dependencies/test_cognito_user_pool_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from secrets import compare_digest
 
 from aws_lambda_powertools.utilities.data_classes.cognito_user_pool_event import (

tests/unit/data_classes/required_dependencies/test_connect_contact_flow_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes.connect_contact_flow_event import (
     ConnectContactFlowChannel,
     ConnectContactFlowEndpointType,

tests/unit/data_classes/required_dependencies/test_dynamo_db_stream_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from decimal import Clamped, Context, Inexact, Overflow, Rounded, Underflow
 
 from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (

tests/unit/data_classes/required_dependencies/test_event_bridge_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import EventBridgeEvent
 from tests.functional.utils import load_event
 

tests/unit/data_classes/required_dependencies/test_iot_registry_events.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from datetime import datetime
 
 from aws_lambda_powertools.utilities.data_classes.iot_registry_event import (

tests/unit/data_classes/required_dependencies/test_kafka_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.utilities.data_classes import KafkaEvent

tests/unit/data_classes/required_dependencies/test_kinesis_firehose_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import KinesisFirehoseEvent
 from tests.functional.utils import load_event
 

tests/unit/data_classes/required_dependencies/test_kinesis_firehose_response.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import (
     KinesisFirehoseDataTransformationRecord,
     KinesisFirehoseDataTransformationRecordMetadata,

tests/unit/data_classes/required_dependencies/test_kinesis_stream_event.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import base64
 import json
 

tests/unit/data_classes/required_dependencies/test_lambda_function_url.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent
 from aws_lambda_powertools.utilities.data_classes.api_gateway_proxy_event import RequestContextV2Authorizer
 from tests.functional.utils import load_event