refactor(examples): add from __future__ import annotations

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

Author: ericbn

examples/batch_processing/src/advanced_accessing_lambda_context.py

@@ -1,21 +1,25 @@
-from typing import Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()
 logger = Logger()
 
 
 @tracer.capture_method
-def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
+def record_handler(record: SQSRecord, lambda_context: LambdaContext | None = None):
     if lambda_context is not None:
         remaining_time = lambda_context.get_remaining_time_in_millis()
         logger.info(remaining_time)

examples/batch_processing/src/advanced_accessing_lambda_context_decorator.py

@@ -1,21 +1,25 @@
-from typing import Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     batch_processor,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()
 logger = Logger()
 
 
 @tracer.capture_method
-def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
+def record_handler(record: SQSRecord, lambda_context: LambdaContext | None = None):
     if lambda_context is not None:
         remaining_time = lambda_context.get_remaining_time_in_millis()
         logger.info(remaining_time)

examples/batch_processing/src/advanced_accessing_lambda_context_manager.py

@@ -1,17 +1,21 @@
-from typing import Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()
 logger = Logger()
 
 
 @tracer.capture_method
-def record_handler(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
+def record_handler(record: SQSRecord, lambda_context: LambdaContext | None = None):
     if lambda_context is not None:
         remaining_time = lambda_context.get_remaining_time_in_millis()
         logger.info(remaining_time)

examples/batch_processing/src/context_manager_access.py

@@ -1,14 +1,16 @@
 from __future__ import annotations
 
 import json
-from typing import List, Tuple
+from typing import TYPE_CHECKING
 
 from typing_extensions import Literal
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()
@@ -28,7 +30,7 @@ def record_handler(record: SQSRecord):
 def lambda_handler(event, context: LambdaContext):
     batch = event["Records"]  # (1)!
     with processor(records=batch, handler=record_handler):
-        processed_messages: List[Tuple] = processor.process()
+        processed_messages: list[tuple] = processor.process()
 
     for message in processed_messages:
         status: Literal["success", "fail"] = message[0]

examples/batch_processing/src/custom_partial_processor.py

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import copy
 import os
 import sys
 from random import randint
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import boto3
 
@@ -11,7 +13,9 @@
     BasePartialProcessor,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.batch.types import PartialItemFailureResponse
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.batch.types import PartialItemFailureResponse
 
 table_name = os.getenv("TABLE_NAME", "table_store_batch")
 

examples/batch_processing/src/disable_tracing.py

@@ -1,13 +1,18 @@
+from __future__ import annotations
+
 import json
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()

examples/batch_processing/src/extending_processor_handlers.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import json
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools import Logger, Metrics, Tracer
 from aws_lambda_powertools.metrics import MetricUnit
@@ -10,13 +12,15 @@
     FailureResponse,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.batch.base import SuccessResponse
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.batch.base import SuccessResponse
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 
 class MyProcessor(BatchProcessor):
-    def success_handler(self, record: Dict[str, Any], result: Any) -> SuccessResponse:
+    def success_handler(self, record: dict[str, Any], result: Any) -> SuccessResponse:
         metrics.add_metric(name="BatchRecordSuccesses", unit=MetricUnit.Count, value=1)
         return super().success_handler(record, result)
 

examples/batch_processing/src/getting_started_async.py

@@ -1,12 +1,18 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import httpx  # external dependency
 
 from aws_lambda_powertools.utilities.batch import (
     AsyncBatchProcessor,
     EventType,
     async_process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = AsyncBatchProcessor(event_type=EventType.SQS)
 

examples/batch_processing/src/getting_started_dynamodb.py

@@ -1,15 +1,20 @@
+from __future__ import annotations
+
 import json
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
-    DynamoDBRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
+        DynamoDBRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.DynamoDBStreams)  # (1)!
 tracer = Tracer()

examples/batch_processing/src/getting_started_dynamodb_context_manager.py

@@ -1,11 +1,16 @@
+from __future__ import annotations
+
 import json
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
-from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
-    DynamoDBRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
+        DynamoDBRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.DynamoDBStreams)
 tracer = Tracer()

examples/batch_processing/src/getting_started_dynamodb_decorator.py

@@ -1,15 +1,20 @@
+from __future__ import annotations
+
 import json
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     batch_processor,
 )
-from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
-    DynamoDBRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
+        DynamoDBRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.DynamoDBStreams)
 tracer = Tracer()

examples/batch_processing/src/getting_started_error_handling.py

@@ -1,19 +1,24 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)
 tracer = Tracer()
 logger = Logger()
 
 
-class InvalidPayload(Exception):
-    ...
+class InvalidPayload(Exception): ...
 
 
 @tracer.capture_method

examples/batch_processing/src/getting_started_kinesis.py

@@ -1,13 +1,19 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
-    KinesisStreamRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
+        KinesisStreamRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.KinesisDataStreams)  # (1)!
 tracer = Tracer()

examples/batch_processing/src/getting_started_kinesis_context_manager.py

@@ -1,9 +1,15 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType
-from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
-    KinesisStreamRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
+        KinesisStreamRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.KinesisDataStreams)
 tracer = Tracer()

examples/batch_processing/src/getting_started_kinesis_decorator.py

@@ -1,13 +1,19 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     batch_processor,
 )
-from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
-    KinesisStreamRecord,
-)
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.kinesis_stream_event import (
+        KinesisStreamRecord,
+    )
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.KinesisDataStreams)
 tracer = Tracer()

examples/batch_processing/src/getting_started_sqs.py

@@ -1,11 +1,17 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.batch import (
     BatchProcessor,
     EventType,
     process_partial_response,
 )
-from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 processor = BatchProcessor(event_type=EventType.SQS)  # (1)!
 tracer = Tracer()