Using generics types + enabling ruff

Author: leandrodamascena

tests/e2e/parser/handlers/handler_with_basic_model.py

@@ -15,6 +15,6 @@ class BasicModel(BaseModel):
     version: str
 
 
-@event_parser
+@event_parser(model=BasicModel)
 def lambda_handler(event: BasicModel, context: LambdaContext):
     return {"product": event.product}

tests/e2e/parser/handlers/handler_with_dataclass.py

@@ -15,6 +15,6 @@ class BasicDataclass:
     version: str
 
 
-@event_parser
+@event_parser(model=BasicDataclass)
 def lambda_handler(event: BasicDataclass, context: LambdaContext):
     return {"product": event.product}

tests/e2e/parser/handlers/handler_with_union_tag.py

@@ -30,6 +30,6 @@ class PartialFailureCallback(BaseModel):
 OrderCallback = Annotated[Union[SuccessCallback, ErrorCallback, PartialFailureCallback], Field(discriminator="status")]
 
 
-@event_parser
+@event_parser(model=OrderCallback)
 def lambda_handler(event: OrderCallback, context: LambdaContext):
     return {"error_msg": event.error_msg}

tests/e2e/utils/data_fetcher/common.py

@@ -1,9 +1,8 @@
-from __future__ import annotations
-
 import functools
 import time
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
+from typing import List, Optional, Tuple
 
 import boto3
 import requests
@@ -14,13 +13,13 @@
 from requests.exceptions import RequestException
 from retry import retry
 
-GetLambdaResponse = tuple[InvocationResponseTypeDef, datetime]
+GetLambdaResponse = Tuple[InvocationResponseTypeDef, datetime]
 
 
 class GetLambdaResponseOptions(BaseModel):
     lambda_arn: str
-    payload: str | None = None
-    client: LambdaClient | None = None
+    payload: Optional[str] = None
+    client: Optional[LambdaClient] = None
     raise_on_error: bool = True
 
     model_config = ConfigDict(
@@ -30,8 +29,8 @@ class GetLambdaResponseOptions(BaseModel):
 
 def get_lambda_response(
     lambda_arn: str,
-    payload: str | None = None,
-    client: LambdaClient | None = None,
+    payload: Optional[str] = None,
+    client: Optional[LambdaClient] = None,
     raise_on_error: bool = True,
 ) -> GetLambdaResponse:
     """Invoke function synchronously
@@ -83,8 +82,8 @@ def get_http_response(request: Request) -> Response:
 
 
 def get_lambda_response_in_parallel(
-    get_lambda_response_options: list[GetLambdaResponseOptions],
-) -> list[GetLambdaResponse]:
+    get_lambda_response_options: List[GetLambdaResponseOptions],
+) -> List[GetLambdaResponse]:
     """Invoke functions in parallel
 
     Parameters
@@ -99,7 +98,7 @@ def get_lambda_response_in_parallel(
     """
     result_list = []
     with ThreadPoolExecutor() as executor:
-        running_tasks: list[Future] = []
+        running_tasks: List[Future] = []
         for options in get_lambda_response_options:
             # Sleep 0.5, 1, 1.5, ... seconds between each invocation. This way
             # we can guarantee that lambdas are executed in parallel, but they are

tests/e2e/utils/data_fetcher/logs.py

@@ -1,39 +1,34 @@
-from __future__ import annotations
-
 import json
-from typing import TYPE_CHECKING
+from datetime import datetime
+from typing import List, Optional, Union
 
 import boto3
+from mypy_boto3_logs.client import CloudWatchLogsClient
 from pydantic import BaseModel
 from retry import retry
 
-if TYPE_CHECKING:
-    from datetime import datetime
-
-    from mypy_boto3_logs.client import CloudWatchLogsClient
-
 
 class Log(BaseModel, extra="allow"):
     level: str
     location: str
-    message: dict | str
+    message: Union[dict, str]
     timestamp: str
     service: str
-    cold_start: bool | None = None
-    function_name: str | None = None
-    function_memory_size: str | None = None
-    function_arn: str | None = None
-    function_request_id: str | None = None
-    xray_trace_id: str | None = None
+    cold_start: Optional[bool] = None
+    function_name: Optional[str] = None
+    function_memory_size: Optional[str] = None
+    function_arn: Optional[str] = None
+    function_request_id: Optional[str] = None
+    xray_trace_id: Optional[str] = None
 
 
 class LogFetcher:
     def __init__(
         self,
         function_name: str,
         start_time: datetime,
-        log_client: CloudWatchLogsClient | None = None,
-        filter_expression: str | None = None,
+        log_client: Optional[CloudWatchLogsClient] = None,
+        filter_expression: Optional[str] = None,
         minimum_log_entries: int = 1,
     ):
         """Fetch and expose Powertools for AWS Lambda (Python) Logger logs from CloudWatch Logs
@@ -57,9 +52,9 @@ def __init__(
         self.filter_expression = filter_expression or "message"  # Logger message key
         self.log_group = f"/aws/lambda/{self.function_name}"
         self.minimum_log_entries = minimum_log_entries
-        self.logs: list[Log] = self._get_logs()
+        self.logs: List[Log] = self._get_logs()
 
-    def get_log(self, key: str, value: any | None = None) -> list[Log]:
+    def get_log(self, key: str, value: Optional[any] = None) -> List[Log]:
         """Get logs based on key or key and value
 
         Parameters
@@ -83,7 +78,7 @@ def get_log(self, key: str, value: any | None = None) -> list[Log]:
                 logs.append(log)
         return logs
 
-    def get_cold_start_log(self) -> list[Log]:
+    def get_cold_start_log(self) -> List[Log]:
         """Get logs where cold start was true
 
         Returns
@@ -103,7 +98,7 @@ def have_keys(self, *keys) -> bool:
         """
         return all(hasattr(log, key) for log in self.logs for key in keys)
 
-    def _get_logs(self) -> list[Log]:
+    def _get_logs(self) -> List[Log]:
         ret = self.log_client.filter_log_events(
             logGroupName=self.log_group,
             startTime=self.start_time,
@@ -137,8 +132,8 @@ def get_logs(
     function_name: str,
     start_time: datetime,
     minimum_log_entries: int = 1,
-    filter_expression: str | None = None,
-    log_client: CloudWatchLogsClient | None = None,
+    filter_expression: Optional[str] = None,
+    log_client: Optional[CloudWatchLogsClient] = None,
 ) -> LogFetcher:
     """_summary_