Fix pydantic not working with Python 3.8

    TypeError: You have a type annotation 'str | None' which makes use
    of newer typing features than are supported in your version of Python.
    To handle this error, you should either remove the use of new syntax
    or install the `eval_type_backport` package.

Author: ericbn

aws_lambda_powertools/utilities/parser/models/alb.py

@@ -1,4 +1,5 @@
-from __future__ import annotations
+# ruff: noqa: FA100
+from typing import Dict, Type, Union
 
 from pydantic import BaseModel
 
@@ -14,8 +15,8 @@ class AlbRequestContext(BaseModel):
 class AlbModel(BaseModel):
     httpMethod: str
     path: str
-    body: str | type[BaseModel]
+    body: Union[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,9 @@
-from __future__ import annotations
-
-from datetime import datetime  # noqa: TCH003
-from typing import Any, Literal
+# ruff: noqa: FA100
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional, Type, Union
 
 from pydantic import BaseModel, model_validator
-from pydantic.networks import IPvAnyNetwork  # noqa: TCH002
+from pydantic.networks import IPvAnyNetwork
 
 
 class ApiGatewayUserCertValidity(BaseModel):
@@ -21,54 +20,54 @@ class ApiGatewayUserCert(BaseModel):
 
 
 class APIGatewayEventIdentity(BaseModel):
-    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
+    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
     # 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: IPvAnyNetwork | Literal["test-invoke-source-ip"]
-    user: str | None = None
-    userAgent: str | None = None
-    userArn: str | None = None
-    clientCert: ApiGatewayUserCert | None = None
+    sourceIp: Union[IPvAnyNetwork, Literal["test-invoke-source-ip"]]
+    user: Optional[str] = None
+    userAgent: Optional[str] = None
+    userArn: Optional[str] = None
+    clientCert: Optional[ApiGatewayUserCert] = None
 
 
 class APIGatewayEventAuthorizer(BaseModel):
-    claims: dict[str, Any] | None = None
-    scopes: list[str] | None = None
+    claims: Optional[Dict[str, Any]] = None
+    scopes: Optional[List[str]] = None
 
 
 class APIGatewayEventRequestContext(BaseModel):
     accountId: str
     apiId: str
-    authorizer: APIGatewayEventAuthorizer | None = None
+    authorizer: Optional[APIGatewayEventAuthorizer] = None
     stage: str
     protocol: str
     identity: APIGatewayEventIdentity
     requestId: str
     requestTime: str
     requestTimeEpoch: datetime
-    resourceId: str | None = None
+    resourceId: Optional[str] = None
     resourcePath: str
-    domainName: str | None = None
-    domainPrefix: str | None = None
-    extendedRequestId: str | None = None
+    domainName: Optional[str] = None
+    domainPrefix: Optional[str] = None
+    extendedRequestId: Optional[str] = None
     httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
     path: str
-    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
+    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
 
     @model_validator(mode="before")
     def check_message_id(cls, values):
@@ -79,16 +78,16 @@ def check_message_id(cls, values):
 
 
 class APIGatewayProxyEventModel(BaseModel):
-    version: str | None = None
+    version: Optional[str] = None
     resource: str
     path: str
     httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
-    headers: dict[str, str]
-    multiValueHeaders: dict[str, list[str]]
-    queryStringParameters: dict[str, str] | None = None
-    multiValueQueryStringParameters: dict[str, list[str]] | None = None
+    headers: Dict[str, str]
+    multiValueHeaders: Dict[str, List[str]]
+    queryStringParameters: Optional[Dict[str, str]] = None
+    multiValueQueryStringParameters: Optional[Dict[str, List[str]]] = None
     requestContext: APIGatewayEventRequestContext
-    pathParameters: dict[str, str] | None = None
-    stageVariables: dict[str, str] | None = None
+    pathParameters: Optional[Dict[str, str]] = None
+    stageVariables: Optional[Dict[str, str]] = None
     isBase64Encoded: bool
-    body: str | type[BaseModel] | None = None
+    body: Optional[Union[str, Type[BaseModel]]] = None

aws_lambda_powertools/utilities/parser/models/apigwv2.py

@@ -1,37 +1,36 @@
-from __future__ import annotations
-
-from datetime import datetime  # noqa: TCH003
-from typing import Any, Literal
+# ruff: noqa: FA100
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional, Type, Union
 
 from pydantic import BaseModel, Field
-from pydantic.networks import IPvAnyNetwork  # noqa: TCH002
+from pydantic.networks import IPvAnyNetwork
 
 
 class RequestContextV2AuthorizerIamCognito(BaseModel):
-    amr: list[str]
+    amr: List[str]
     identityId: str
     identityPoolId: str
 
 
 class RequestContextV2AuthorizerIam(BaseModel):
-    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
+    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
 
 
 class RequestContextV2AuthorizerJwt(BaseModel):
-    claims: dict[str, Any]
-    scopes: list[str] | None = None
+    claims: Dict[str, Any]
+    scopes: Optional[List[str]] = None
 
 
 class RequestContextV2Authorizer(BaseModel):
-    jwt: RequestContextV2AuthorizerJwt | None = None
-    iam: RequestContextV2AuthorizerIam | None = None
-    lambda_value: dict[str, Any] | None = Field(None, alias="lambda")
+    jwt: Optional[RequestContextV2AuthorizerJwt] = None
+    iam: Optional[RequestContextV2AuthorizerIam] = None
+    lambda_value: Optional[Dict[str, Any]] = Field(None, alias="lambda")
 
 
 class RequestContextV2Http(BaseModel):
@@ -45,7 +44,7 @@ class RequestContextV2Http(BaseModel):
 class RequestContextV2(BaseModel):
     accountId: str
     apiId: str
-    authorizer: RequestContextV2Authorizer | None = None
+    authorizer: Optional[RequestContextV2Authorizer] = None
     domainName: str
     domainPrefix: str
     requestId: str
@@ -61,11 +60,11 @@ class APIGatewayProxyEventV2Model(BaseModel):
     routeKey: str
     rawPath: str
     rawQueryString: str
-    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
+    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
     requestContext: RequestContextV2
-    body: str | type[BaseModel] | None = None
+    body: Optional[Union[str, Type[BaseModel]]] = None
     isBase64Encoded: bool

aws_lambda_powertools/utilities/parser/models/bedrock_agent.py

@@ -1,4 +1,5 @@
-from __future__ import annotations
+# ruff: noqa: FA100
+from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field
 
@@ -17,11 +18,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 +32,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: list[BedrockAgentPropertyModel] | None = None
-    request_body: BedrockAgentRequestBodyModel | None = Field(None, alias="requestBody")
+    parameters: Optional[List[BedrockAgentPropertyModel]] = None
+    request_body: Optional[BedrockAgentRequestBodyModel] = Field(None, alias="requestBody")

aws_lambda_powertools/utilities/parser/models/cloudformation_custom_resource.py

@@ -1,6 +1,5 @@
-from __future__ import annotations
-
-from typing import Any, Literal
+# ruff: noqa: FA100
+from typing import Any, Dict, Literal, Union
 
 from pydantic import BaseModel, Field, HttpUrl
 
@@ -13,7 +12,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: dict[str, Any] | BaseModel | None = Field(None, alias="ResourceProperties")
+    resource_properties: Union[Dict[str, Any], BaseModel, None] = Field(None, alias="ResourceProperties")
 
 
 class CloudFormationCustomResourceCreateModel(CloudFormationCustomResourceBaseModel):
@@ -28,4 +27,4 @@ class CloudFormationCustomResourceDeleteModel(CloudFormationCustomResourceBaseMo
 class CloudFormationCustomResourceUpdateModel(CloudFormationCustomResourceBaseModel):
     request_type: Literal["Update"] = Field(..., alias="RequestType")
     physical_resource_id: str = Field(..., alias="PhysicalResourceId")
-    old_resource_properties: dict[str, Any] | BaseModel | None = Field(None, alias="OldResourceProperties")
+    old_resource_properties: Union[Dict[str, Any], BaseModel, None] = Field(None, alias="OldResourceProperties")

aws_lambda_powertools/utilities/parser/models/cloudwatch.py

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

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -1,7 +1,6 @@
-from __future__ import annotations
-
-from datetime import datetime  # noqa: TCH003
-from typing import Any, Literal
+# ruff: noqa: FA100
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional, Type, Union
 
 from pydantic import BaseModel, field_validator
 
@@ -11,10 +10,10 @@
 
 
 class DynamoDBStreamChangedRecordModel(BaseModel):
-    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
+    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
     SequenceNumber: str
     SizeBytes: int
     StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]
@@ -48,8 +47,8 @@ class DynamoDBStreamRecordModel(BaseModel):
     awsRegion: str
     eventSourceARN: str
     dynamodb: DynamoDBStreamChangedRecordModel
-    userIdentity: UserIdentity | None = None
+    userIdentity: Optional[UserIdentity] = None
 
 
 class DynamoDBStreamModel(BaseModel):
-    Records: list[DynamoDBStreamRecordModel]
+    Records: List[DynamoDBStreamRecordModel]

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,10 +1,10 @@
-from __future__ import annotations
-
-from datetime import datetime  # noqa: TCH003
+# ruff: noqa: FA100
+from datetime import datetime
+from typing import List, Optional
 
 from pydantic import BaseModel, Field
 
-from aws_lambda_powertools.utilities.parser.types import RawDictOrModel  # noqa: TCH001
+from aws_lambda_powertools.utilities.parser.types import RawDictOrModel
 
 
 class EventBridgeModel(BaseModel):
@@ -14,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: str | None = Field(None, alias="replay-name")
+    replay_name: Optional[str] = Field(None, alias="replay-name")