refactor(parser): add from __future__ import annotations

aws_lambda_powertools/utilities/parser/functions.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from pydantic import TypeAdapter
 
 from aws_lambda_powertools.shared.cache_dict import LRUDict
-from aws_lambda_powertools.utilities.parser.types import T
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parser.types import T
 
 CACHE_TYPE_ADAPTER = LRUDict(max_items=1024)
 

aws_lambda_powertools/utilities/parser/models/alb.py

@@ -1,4 +1,4 @@
-from typing import Dict, Type, Union
+from __future__ import annotations
 
 from pydantic import BaseModel
 
@@ -14,8 +14,8 @@ class AlbRequestContext(BaseModel):
 class AlbModel(BaseModel):
     httpMethod: str
     path: str
-    body: Union[str, Type[BaseModel]]
+    body: str | type[BaseModel]
     isBase64Encoded: bool
-    headers: Dict[str, str]
-    queryStringParameters: Dict[str, str]
+    headers: dict[str, str]
+    queryStringParameters: dict[str, str]
     requestContext: AlbRequestContext

aws_lambda_powertools/utilities/parser/models/apigw.py

@@ -1,10 +1,10 @@
-from datetime import datetime
-from typing import Any, Dict, List, Optional, Type, Union
+from __future__ import annotations
 
-from pydantic import BaseModel, model_validator
-from pydantic.networks import IPvAnyNetwork
+from datetime import datetime  # noqa: TCH003
+from typing import Any, Literal
 
-from aws_lambda_powertools.utilities.parser.types import Literal
+from pydantic import BaseModel, model_validator
+from pydantic.networks import IPvAnyNetwork  # noqa: TCH002
 
 
 class ApiGatewayUserCertValidity(BaseModel):
@@ -21,54 +21,54 @@ class ApiGatewayUserCert(BaseModel):
 
 
 class APIGatewayEventIdentity(BaseModel):
-    accessKey: Optional[str] = None
-    accountId: Optional[str] = None
-    apiKey: Optional[str] = None
-    apiKeyId: Optional[str] = None
-    caller: Optional[str] = None
-    cognitoAuthenticationProvider: Optional[str] = None
-    cognitoAuthenticationType: Optional[str] = None
-    cognitoIdentityId: Optional[str] = None
-    cognitoIdentityPoolId: Optional[str] = None
-    principalOrgId: Optional[str] = None
+    accessKey: str | None = None
+    accountId: str | None = None
+    apiKey: str | None = None
+    apiKeyId: str | None = None
+    caller: str | None = None
+    cognitoAuthenticationProvider: str | None = None
+    cognitoAuthenticationType: str | None = None
+    cognitoIdentityId: str | None = None
+    cognitoIdentityPoolId: str | None = None
+    principalOrgId: str | None = None
     # see #1562, temp workaround until API Gateway fixes it the Test button payload
     # removing it will not be considered a regression in the future
-    sourceIp: Union[IPvAnyNetwork, Literal["test-invoke-source-ip"]]
-    user: Optional[str] = None
-    userAgent: Optional[str] = None
-    userArn: Optional[str] = None
-    clientCert: Optional[ApiGatewayUserCert] = None
+    sourceIp: IPvAnyNetwork | Literal["test-invoke-source-ip"]
+    user: str | None = None
+    userAgent: str | None = None
+    userArn: str | None = None
+    clientCert: ApiGatewayUserCert | None = None
 
 
 class APIGatewayEventAuthorizer(BaseModel):
-    claims: Optional[Dict[str, Any]] = None
-    scopes: Optional[List[str]] = None
+    claims: dict[str, Any] | None = None
+    scopes: list[str] | None = None
 
 
 class APIGatewayEventRequestContext(BaseModel):
     accountId: str
     apiId: str
-    authorizer: Optional[APIGatewayEventAuthorizer] = None
+    authorizer: APIGatewayEventAuthorizer | None = None
     stage: str
     protocol: str
     identity: APIGatewayEventIdentity
     requestId: str
     requestTime: str
     requestTimeEpoch: datetime
-    resourceId: Optional[str] = None
+    resourceId: str | None = None
     resourcePath: str
-    domainName: Optional[str] = None
-    domainPrefix: Optional[str] = None
-    extendedRequestId: Optional[str] = None
+    domainName: str | None = None
+    domainPrefix: str | None = None
+    extendedRequestId: str | None = None
     httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
     path: str
-    connectedAt: Optional[datetime] = None
-    connectionId: Optional[str] = None
-    eventType: Optional[Literal["CONNECT", "MESSAGE", "DISCONNECT"]] = None
-    messageDirection: Optional[str] = None
-    messageId: Optional[str] = None
-    routeKey: Optional[str] = None
-    operationName: Optional[str] = None
+    connectedAt: datetime | None = None
+    connectionId: str | None = None
+    eventType: Literal["CONNECT", "MESSAGE", "DISCONNECT"] | None = None
+    messageDirection: str | None = None
+    messageId: str | None = None
+    routeKey: str | None = None
+    operationName: str | None = None
 
     @model_validator(mode="before")
     def check_message_id(cls, values):
@@ -79,16 +79,16 @@ def check_message_id(cls, values):
 
 
 class APIGatewayProxyEventModel(BaseModel):
-    version: Optional[str] = None
+    version: str | None = None
     resource: str
     path: str
     httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
-    headers: Dict[str, str]
-    multiValueHeaders: Dict[str, List[str]]
-    queryStringParameters: Optional[Dict[str, str]] = None
-    multiValueQueryStringParameters: Optional[Dict[str, List[str]]] = None
+    headers: dict[str, str]
+    multiValueHeaders: dict[str, list[str]]
+    queryStringParameters: dict[str, str] | None = None
+    multiValueQueryStringParameters: dict[str, list[str]] | None = None
     requestContext: APIGatewayEventRequestContext
-    pathParameters: Optional[Dict[str, str]] = None
-    stageVariables: Optional[Dict[str, str]] = None
+    pathParameters: dict[str, str] | None = None
+    stageVariables: dict[str, str] | None = None
     isBase64Encoded: bool
-    body: Optional[Union[str, Type[BaseModel]]] = None
+    body: str | type[BaseModel] | None = None

aws_lambda_powertools/utilities/parser/models/apigwv2.py

@@ -1,37 +1,37 @@
-from datetime import datetime
-from typing import Any, Dict, List, Optional, Type, Union
+from __future__ import annotations
 
-from pydantic import BaseModel, Field
-from pydantic.networks import IPvAnyNetwork
+from datetime import datetime  # noqa: TCH003
+from typing import Any, Literal
 
-from aws_lambda_powertools.utilities.parser.types import Literal
+from pydantic import BaseModel, Field
+from pydantic.networks import IPvAnyNetwork  # noqa: TCH002
 
 
 class RequestContextV2AuthorizerIamCognito(BaseModel):
-    amr: List[str]
+    amr: list[str]
     identityId: str
     identityPoolId: str
 
 
 class RequestContextV2AuthorizerIam(BaseModel):
-    accessKey: Optional[str] = None
-    accountId: Optional[str] = None
-    callerId: Optional[str] = None
-    principalOrgId: Optional[str] = None
-    userArn: Optional[str] = None
-    userId: Optional[str] = None
-    cognitoIdentity: Optional[RequestContextV2AuthorizerIamCognito] = None
+    accessKey: str | None = None
+    accountId: str | None = None
+    callerId: str | None = None
+    principalOrgId: str | None = None
+    userArn: str | None = None
+    userId: str | None = None
+    cognitoIdentity: RequestContextV2AuthorizerIamCognito | None = None
 
 
 class RequestContextV2AuthorizerJwt(BaseModel):
-    claims: Dict[str, Any]
-    scopes: Optional[List[str]] = None
+    claims: dict[str, Any]
+    scopes: list[str] | None = None
 
 
 class RequestContextV2Authorizer(BaseModel):
-    jwt: Optional[RequestContextV2AuthorizerJwt] = None
-    iam: Optional[RequestContextV2AuthorizerIam] = None
-    lambda_value: Optional[Dict[str, Any]] = Field(None, alias="lambda")
+    jwt: RequestContextV2AuthorizerJwt | None = None
+    iam: RequestContextV2AuthorizerIam | None = None
+    lambda_value: dict[str, Any] | None = Field(None, alias="lambda")
 
 
 class RequestContextV2Http(BaseModel):
@@ -45,7 +45,7 @@ class RequestContextV2Http(BaseModel):
 class RequestContextV2(BaseModel):
     accountId: str
     apiId: str
-    authorizer: Optional[RequestContextV2Authorizer] = None
+    authorizer: RequestContextV2Authorizer | None = None
     domainName: str
     domainPrefix: str
     requestId: str
@@ -61,11 +61,11 @@ class APIGatewayProxyEventV2Model(BaseModel):
     routeKey: str
     rawPath: str
     rawQueryString: str
-    cookies: Optional[List[str]] = None
-    headers: Dict[str, str]
-    queryStringParameters: Optional[Dict[str, str]] = None
-    pathParameters: Optional[Dict[str, str]] = None
-    stageVariables: Optional[Dict[str, str]] = None
+    cookies: list[str] | None = None
+    headers: dict[str, str]
+    queryStringParameters: dict[str, str] | None = None
+    pathParameters: dict[str, str] | None = None
+    stageVariables: dict[str, str] | None = None
     requestContext: RequestContextV2
-    body: Optional[Union[str, Type[BaseModel]]] = None
+    body: str | type[BaseModel] | None = None
     isBase64Encoded: bool

aws_lambda_powertools/utilities/parser/models/bedrock_agent.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional
+from __future__ import annotations
 
 from pydantic import BaseModel, Field
 
@@ -17,11 +17,11 @@ class BedrockAgentPropertyModel(BaseModel):
 
 
 class BedrockAgentRequestMediaModel(BaseModel):
-    properties: List[BedrockAgentPropertyModel]
+    properties: list[BedrockAgentPropertyModel]
 
 
 class BedrockAgentRequestBodyModel(BaseModel):
-    content: Dict[str, BedrockAgentRequestMediaModel]
+    content: dict[str, BedrockAgentRequestMediaModel]
 
 
 class BedrockAgentEventModel(BaseModel):
@@ -31,8 +31,8 @@ class BedrockAgentEventModel(BaseModel):
     action_group: str = Field(..., alias="actionGroup")
     api_path: str = Field(..., alias="apiPath")
     http_method: str = Field(..., alias="httpMethod")
-    session_attributes: Dict[str, str] = Field({}, alias="sessionAttributes")
-    prompt_session_attributes: Dict[str, str] = Field({}, alias="promptSessionAttributes")
+    session_attributes: dict[str, str] = Field({}, alias="sessionAttributes")
+    prompt_session_attributes: dict[str, str] = Field({}, alias="promptSessionAttributes")
     agent: BedrockAgentModel
-    parameters: Optional[List[BedrockAgentPropertyModel]] = None
-    request_body: Optional[BedrockAgentRequestBodyModel] = Field(None, alias="requestBody")
+    parameters: list[BedrockAgentPropertyModel] | None = None
+    request_body: BedrockAgentRequestBodyModel | None = Field(None, alias="requestBody")

aws_lambda_powertools/utilities/parser/models/cloudformation_custom_resource.py

@@ -1,8 +1,8 @@
-from typing import Any, Dict, Union
+from __future__ import annotations
 
-from pydantic import BaseModel, Field, HttpUrl
+from typing import Any, Literal
 
-from aws_lambda_powertools.utilities.parser.types import Literal
+from pydantic import BaseModel, Field, HttpUrl
 
 
 class CloudFormationCustomResourceBaseModel(BaseModel):
@@ -13,7 +13,7 @@ class CloudFormationCustomResourceBaseModel(BaseModel):
     request_id: str = Field(..., alias="RequestId")
     logical_resource_id: str = Field(..., alias="LogicalResourceId")
     resource_type: str = Field(..., alias="ResourceType")
-    resource_properties: Union[Dict[str, Any], BaseModel, None] = Field(None, alias="ResourceProperties")
+    resource_properties: dict[str, Any] | BaseModel | None = Field(None, alias="ResourceProperties")
 
 
 class CloudFormationCustomResourceCreateModel(CloudFormationCustomResourceBaseModel):
@@ -28,4 +28,4 @@ class CloudFormationCustomResourceDeleteModel(CloudFormationCustomResourceBaseMo
 class CloudFormationCustomResourceUpdateModel(CloudFormationCustomResourceBaseModel):
     request_type: Literal["Update"] = Field(..., alias="RequestType")
     physical_resource_id: str = Field(..., alias="PhysicalResourceId")
-    old_resource_properties: Union[Dict[str, Any], BaseModel, None] = Field(None, alias="OldResourceProperties")
+    old_resource_properties: dict[str, Any] | BaseModel | None = Field(None, alias="OldResourceProperties")

aws_lambda_powertools/utilities/parser/models/cloudwatch.py

@@ -1,9 +1,10 @@
+from __future__ import annotations
+
 import base64
 import json
 import logging
 import zlib
-from datetime import datetime
-from typing import List, Optional, Type, Union
+from datetime import datetime  # noqa: TCH003
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -13,17 +14,17 @@
 class CloudWatchLogsLogEvent(BaseModel):
     id: str  # noqa AA03 VNE003
     timestamp: datetime
-    message: Union[str, Type[BaseModel]]
+    message: str | type[BaseModel]
 
 
 class CloudWatchLogsDecode(BaseModel):
     messageType: str
     owner: str
     logGroup: str
     logStream: str
-    subscriptionFilters: List[str]
-    logEvents: List[CloudWatchLogsLogEvent]
-    policyLevel: Optional[str] = None
+    subscriptionFilters: list[str]
+    logEvents: list[CloudWatchLogsLogEvent]
+    policyLevel: str | None = None
 
 
 class CloudWatchLogsData(BaseModel):

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -1,19 +1,20 @@
-from datetime import datetime
-from typing import Any, Dict, List, Optional, Type, Union
+from __future__ import annotations
+
+from datetime import datetime  # noqa: TCH003
+from typing import Any, Literal
 
 from pydantic import BaseModel, field_validator
 
 from aws_lambda_powertools.shared.dynamodb_deserializer import TypeDeserializer
-from aws_lambda_powertools.utilities.parser.types import Literal
 
 _DESERIALIZER = TypeDeserializer()
 
 
 class DynamoDBStreamChangedRecordModel(BaseModel):
-    ApproximateCreationDateTime: Optional[datetime] = None
-    Keys: Dict[str, Any]
-    NewImage: Optional[Union[Dict[str, Any], Type[BaseModel], BaseModel]] = None
-    OldImage: Optional[Union[Dict[str, Any], Type[BaseModel], BaseModel]] = None
+    ApproximateCreationDateTime: datetime | None = None
+    Keys: dict[str, Any]
+    NewImage: dict[str, Any] | type[BaseModel] | BaseModel | None = None
+    OldImage: dict[str, Any] | type[BaseModel] | BaseModel | None = None
     SequenceNumber: str
     SizeBytes: int
     StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]
@@ -47,8 +48,8 @@ class DynamoDBStreamRecordModel(BaseModel):
     awsRegion: str
     eventSourceARN: str
     dynamodb: DynamoDBStreamChangedRecordModel
-    userIdentity: Optional[UserIdentity] = None
+    userIdentity: UserIdentity | None = None
 
 
 class DynamoDBStreamModel(BaseModel):
-    Records: List[DynamoDBStreamRecordModel]
+    Records: list[DynamoDBStreamRecordModel]

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,9 +1,10 @@
-from datetime import datetime
-from typing import List, Optional
+from __future__ import annotations
+
+from datetime import datetime  # noqa: TCH003
 
 from pydantic import BaseModel, Field
 
-from aws_lambda_powertools.utilities.parser.types import RawDictOrModel
+from aws_lambda_powertools.utilities.parser.types import RawDictOrModel  # noqa: TCH001
 
 
 class EventBridgeModel(BaseModel):
@@ -13,7 +14,7 @@ class EventBridgeModel(BaseModel):
     account: str
     time: datetime
     region: str
-    resources: List[str]
+    resources: list[str]
     detail_type: str = Field(None, alias="detail-type")
     detail: RawDictOrModel
-    replay_name: Optional[str] = Field(None, alias="replay-name")
+    replay_name: str | None = Field(None, alias="replay-name")