refactor(data_classes): add from __future__ import annotations

aws_lambda_powertools/utilities/data_classes/active_mq_event.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 from functools import cached_property
-from typing import Any, Dict, Iterator, Optional
+from typing import Any, Iterator
 
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 from aws_lambda_powertools.utilities.data_classes.shared_functions import base64_decode
@@ -62,32 +64,32 @@ def destination_physicalname(self) -> str:
         return self["destination"]["physicalName"]
 
     @property
-    def delivery_mode(self) -> Optional[int]:
+    def delivery_mode(self) -> int | None:
         """persistent or non-persistent delivery"""
         return self.get("deliveryMode")
 
     @property
-    def correlation_id(self) -> Optional[str]:
+    def correlation_id(self) -> str | None:
         """User defined correlation id"""
         return self.get("correlationID")
 
     @property
-    def reply_to(self) -> Optional[str]:
+    def reply_to(self) -> str | None:
         """User defined reply to"""
         return self.get("replyTo")
 
     @property
-    def get_type(self) -> Optional[str]:
+    def get_type(self) -> str | None:
         """User defined message type"""
         return self.get("type")
 
     @property
-    def expiration(self) -> Optional[int]:
+    def expiration(self) -> int | None:
         """Expiration attribute whose value is given in milliseconds"""
         return self.get("expiration")
 
     @property
-    def priority(self) -> Optional[int]:
+    def priority(self) -> int | None:
         """
         JMS defines a ten-level priority value, with 0 as the lowest priority and 9
         as the highest. In addition, clients should consider priorities 0-4 as
@@ -110,9 +112,9 @@ class ActiveMQEvent(DictWrapper):
     - https://aws.amazon.com/blogs/compute/using-amazon-mq-as-an-event-source-for-aws-lambda/
     """
 
-    def __init__(self, data: Dict[str, Any]):
+    def __init__(self, data: dict[str, Any]):
         super().__init__(data)
-        self._messages: Optional[Iterator[ActiveMQMessage]] = None
+        self._messages: Iterator[ActiveMQMessage] | None = None
 
     @property
     def event_source(self) -> str:

aws_lambda_powertools/utilities/data_classes/alb_event.py

@@ -1,4 +1,6 @@
-from typing import Any, Dict, List
+from __future__ import annotations
+
+from typing import Any
 
 from aws_lambda_powertools.shared.headers_serializer import (
     BaseHeadersSerializer,
@@ -33,19 +35,19 @@ def request_context(self) -> ALBEventRequestContext:
         return ALBEventRequestContext(self._data)
 
     @property
-    def multi_value_query_string_parameters(self) -> Dict[str, List[str]]:
+    def multi_value_query_string_parameters(self) -> dict[str, list[str]]:
         return self.get("multiValueQueryStringParameters") or {}
 
     @property
-    def resolved_query_string_parameters(self) -> Dict[str, List[str]]:
+    def resolved_query_string_parameters(self) -> dict[str, list[str]]:
         return self.multi_value_query_string_parameters or super().resolved_query_string_parameters
 
     @property
-    def multi_value_headers(self) -> Dict[str, List[str]]:
+    def multi_value_headers(self) -> dict[str, list[str]]:
         return CaseInsensitiveDict(self.get("multiValueHeaders"))
 
     @property
-    def resolved_headers_field(self) -> Dict[str, Any]:
+    def resolved_headers_field(self) -> dict[str, Any]:
         return self.multi_value_headers or self.headers
 
     def header_serializer(self) -> BaseHeadersSerializer:

aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import enum
 import re
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from aws_lambda_powertools.utilities.data_classes.common import (
     BaseRequestContext,
@@ -141,19 +143,19 @@ def http_method(self) -> str:
         return self["httpMethod"]
 
     @property
-    def headers(self) -> Dict[str, str]:
+    def headers(self) -> dict[str, str]:
         return CaseInsensitiveDict(self["headers"])
 
     @property
-    def query_string_parameters(self) -> Dict[str, str]:
+    def query_string_parameters(self) -> dict[str, str]:
         return self["queryStringParameters"]
 
     @property
-    def path_parameters(self) -> Dict[str, str]:
+    def path_parameters(self) -> dict[str, str]:
         return self["pathParameters"]
 
     @property
-    def stage_variables(self) -> Dict[str, str]:
+    def stage_variables(self) -> dict[str, str]:
         return self["stageVariables"]
 
     @property
@@ -193,7 +195,7 @@ def parsed_arn(self) -> APIGatewayRouteArn:
         return parse_api_gateway_arn(self.route_arn)
 
     @property
-    def identity_source(self) -> List[str]:
+    def identity_source(self) -> list[str]:
         """The identity source for which authorization is requested.
 
         For a REQUEST authorizer, this is optional. The value is a set of one or more mapping expressions of the
@@ -217,29 +219,29 @@ def raw_query_string(self) -> str:
         return self["rawQueryString"]
 
     @property
-    def cookies(self) -> List[str]:
+    def cookies(self) -> list[str]:
         """Cookies"""
         return self["cookies"]
 
     @property
-    def headers(self) -> Dict[str, str]:
+    def headers(self) -> dict[str, str]:
         """Http headers"""
         return CaseInsensitiveDict(self["headers"])
 
     @property
-    def query_string_parameters(self) -> Dict[str, str]:
+    def query_string_parameters(self) -> dict[str, str]:
         return self["queryStringParameters"]
 
     @property
     def request_context(self) -> BaseRequestContextV2:
         return BaseRequestContextV2(self._data)
 
     @property
-    def path_parameters(self) -> Dict[str, str]:
+    def path_parameters(self) -> dict[str, str]:
         return self.get("pathParameters") or {}
 
     @property
-    def stage_variables(self) -> Dict[str, str]:
+    def stage_variables(self) -> dict[str, str]:
         return self.get("stageVariables") or {}
 
 
@@ -253,7 +255,7 @@ class APIGatewayAuthorizerResponseV2:
         is authorized to make calls to the GraphQL API. If this value is
         true, execution of the GraphQL API continues. If this value is false,
         an UnauthorizedException is raised
-    context: Dict[str, Any], optional
+    context: dict[str, Any], optional
         A JSON object visible as `event.requestContext.authorizer` lambda event
 
         The context object only supports key-value pairs. Nested keys are not supported.
@@ -264,14 +266,14 @@ class APIGatewayAuthorizerResponseV2:
     def __init__(
         self,
         authorize: bool = False,
-        context: Optional[Dict[str, Any]] = None,
+        context: dict[str, Any] | None = None,
     ):
         self.authorize = authorize
         self.context = context
 
     def asdict(self) -> dict:
         """Return the response as a dict"""
-        response: Dict = {"isAuthorized": self.authorize}
+        response: dict = {"isAuthorized": self.authorize}
 
         if self.context:
             response["context"] = self.context
@@ -329,8 +331,8 @@ def __init__(
         aws_account_id: str,
         api_id: str,
         stage: str,
-        context: Optional[Dict] = None,
-        usage_identifier_key: Optional[str] = None,
+        context: dict | None = None,
+        usage_identifier_key: str | None = None,
         partition: str = "aws",
     ):
         """
@@ -357,7 +359,7 @@ def __init__(
             greedily expand over '/' or other separators.
             See https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_resource.html for more
             details.
-        context : Dict, optional
+        context : dict, optional
             Optional, context.
             Note: only names of type string and values of type int, string or boolean are supported
         usage_identifier_key: str, optional
@@ -375,18 +377,18 @@ def __init__(
         self.stage = stage
         self.context = context
         self.usage_identifier_key = usage_identifier_key
-        self._allow_routes: List[Dict] = []
-        self._deny_routes: List[Dict] = []
+        self._allow_routes: list[dict] = []
+        self._deny_routes: list[dict] = []
         self._resource_pattern = re.compile(self.path_regex)
         self.partition = partition
 
     @staticmethod
     def from_route_arn(
         arn: str,
         principal_id: str,
-        context: Optional[Dict] = None,
-        usage_identifier_key: Optional[str] = None,
-    ) -> "APIGatewayAuthorizerResponse":
+        context: dict | None = None,
+        usage_identifier_key: str | None = None,
+    ) -> APIGatewayAuthorizerResponse:
         parsed_arn = parse_api_gateway_arn(arn)
         return APIGatewayAuthorizerResponse(
             principal_id,
@@ -398,7 +400,7 @@ def from_route_arn(
             usage_identifier_key,
         )
 
-    def _add_route(self, effect: str, http_method: str, resource: str, conditions: Optional[List[Dict]] = None):
+    def _add_route(self, effect: str, http_method: str, resource: str, conditions: list[dict] | None = None):
         """Adds a route to the internal lists of allowed or denied routes. Each object in
         the internal list contains a resource ARN and a condition statement. The condition
         statement can be null."""
@@ -427,17 +429,17 @@ def _add_route(self, effect: str, http_method: str, resource: str, conditions: O
             self._deny_routes.append(route)
 
     @staticmethod
-    def _get_empty_statement(effect: str) -> Dict[str, Any]:
+    def _get_empty_statement(effect: str) -> dict[str, Any]:
         """Returns an empty statement object prepopulated with the correct action and the desired effect."""
         return {"Action": "execute-api:Invoke", "Effect": effect.capitalize(), "Resource": []}
 
-    def _get_statement_for_effect(self, effect: str, routes: List[Dict]) -> List[Dict]:
+    def _get_statement_for_effect(self, effect: str, routes: list[dict]) -> list[dict]:
         """This function loops over an array of objects containing a `resourceArn` and
         `conditions` statement and generates the array of statements for the policy."""
         if not routes:
             return []
 
-        statements: List[Dict] = []
+        statements: list[dict] = []
         statement = self._get_empty_statement(effect)
 
         for route in routes:
@@ -476,31 +478,31 @@ def deny_all_routes(self, http_method: str = HttpVerb.ALL.value):
 
         self._add_route(effect="Deny", http_method=http_method, resource="*")
 
-    def allow_route(self, http_method: str, resource: str, conditions: Optional[List[Dict]] = None):
+    def allow_route(self, http_method: str, resource: str, conditions: list[dict] | None = None):
         """Adds an API Gateway method (Http verb + Resource path) to the list of allowed
         methods for the policy.
 
         Optionally includes a condition for the policy statement. More on AWS policy
         conditions here: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition"""
         self._add_route(effect="Allow", http_method=http_method, resource=resource, conditions=conditions)
 
-    def deny_route(self, http_method: str, resource: str, conditions: Optional[List[Dict]] = None):
+    def deny_route(self, http_method: str, resource: str, conditions: list[dict] | None = None):
         """Adds an API Gateway method (Http verb + Resource path) to the list of denied
         methods for the policy.
 
         Optionally includes a condition for the policy statement. More on AWS policy
         conditions here: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html#Condition"""
         self._add_route(effect="Deny", http_method=http_method, resource=resource, conditions=conditions)
 
-    def asdict(self) -> Dict[str, Any]:
+    def asdict(self) -> dict[str, Any]:
         """Generates the policy document based on the internal lists of allowed and denied
         conditions. This will generate a policy with two main statements for the effect:
         one statement for Allow and one statement for Deny.
         Methods that includes conditions will have their own statement in the policy."""
         if len(self._allow_routes) == 0 and len(self._deny_routes) == 0:
             raise ValueError("No statements defined for the policy")
 
-        response: Dict[str, Any] = {
+        response: dict[str, Any] = {
             "principalId": self.principal_id,
             "policyDocument": {"Version": "2012-10-17", "Statement": []},
         }