refactor(typing): use PEP 604 syntax for Optional annotations (UP045)

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -6,9 +6,9 @@
 
 import traceback
 from types import TracebackType
-from typing import Optional, Tuple, Type
+from typing import Tuple, Type
 
-ExceptionInfo = Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]]
+ExceptionInfo = Tuple[Type[BaseException] | None, BaseException | None, TracebackType | None]
 
 
 class BaseBatchProcessingError(Exception):

aws_lambda_powertools/utilities/batch/types.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import sys
-from typing import Optional, Type, TypedDict, Union
+from typing import Type, TypedDict, Union
 
 has_pydantic = "pydantic" in sys.modules
 
@@ -14,15 +14,16 @@
     )
     from aws_lambda_powertools.utilities.parser.models.kafka import KafkaRecordModel
 
-    BatchTypeModels = Optional[
+    BatchTypeModels = (
         Union[
             Type[SqsRecordModel],
             Type[DynamoDBStreamRecordModel],
             Type[KinesisDataStreamRecordModel],
             Type[KafkaRecordModel],
         ]
-    ]
-    BatchSqsTypeModel = Optional[Type[SqsRecordModel]]
+        | None
+    )
+    BatchSqsTypeModel = Type[SqsRecordModel] | None
 else:  # pragma: no cover
     BatchTypeModels = "BatchTypeModels"  # type: ignore
     BatchSqsTypeModel = "BatchSqsTypeModel"  # type: ignore

aws_lambda_powertools/utilities/parser/models/apigw.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Literal, Optional, Type, Union
+from typing import Any, Dict, List, Literal, Type, Union
 
 from pydantic import BaseModel, field_validator, model_validator
 from pydantic.networks import IPvAnyNetwork
@@ -21,23 +21,23 @@ 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, str]
-    user: Optional[str] = None
-    userAgent: Optional[str] = None
-    userArn: Optional[str] = None
-    clientCert: Optional[ApiGatewayUserCert] = None
+    user: str | None = None
+    userAgent: str | None = None
+    userArn: str | None = None
+    clientCert: ApiGatewayUserCert | None = None
 
     @field_validator("sourceIp", mode="before")
     @classmethod
@@ -46,34 +46,34 @@ def _validate_source_ip(cls, value):
 
 
 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):
@@ -84,19 +84,19 @@ 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
+    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
-    isBase64Encoded: Optional[bool] = None
-    body: Optional[Union[str, Type[BaseModel]]] = None
+    pathParameters: Dict[str, str] | None = None
+    stageVariables: Dict[str, str] | None = None
+    isBase64Encoded: bool | None = None
+    body: Union[str, Type[BaseModel]] | None = None
 
 
 class ApiGatewayAuthorizerToken(BaseModel):

aws_lambda_powertools/utilities/parser/models/apigw_websocket.py

@@ -1,13 +1,13 @@
 from datetime import datetime
-from typing import Dict, List, Literal, Optional, Type, Union
+from typing import Dict, List, Literal, Type, Union
 
 from pydantic import BaseModel, Field
 from pydantic.networks import IPvAnyNetwork
 
 
 class APIGatewayWebSocketEventIdentity(BaseModel):
     source_ip: IPvAnyNetwork = Field(alias="sourceIp")
-    user_agent: Optional[str] = Field(None, alias="userAgent")
+    user_agent: str | None = Field(None, alias="userAgent")
 
 
 class APIGatewayWebSocketEventRequestContextBase(BaseModel):
@@ -61,4 +61,4 @@ class APIGatewayWebSocketDisconnectEventModel(BaseModel):
 class APIGatewayWebSocketMessageEventModel(BaseModel):
     request_context: APIGatewayWebSocketMessageEventRequestContext = Field(alias="requestContext")
     is_base64_encoded: bool = Field(alias="isBase64Encoded")
-    body: Optional[Union[str, Type[BaseModel]]] = Field(None, alias="body")
+    body: Union[str, Type[BaseModel]] | None = Field(None, alias="body")

aws_lambda_powertools/utilities/parser/models/apigwv2.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Literal, Optional, Type, Union
+from typing import Any, Dict, List, Literal, Type, Union
 
 from pydantic import BaseModel, Field, field_validator
 from pydantic.networks import IPvAnyNetwork
@@ -14,24 +14,24 @@ class RequestContextV2AuthorizerIamCognito(BaseModel):
 
 
 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
+    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):
@@ -50,7 +50,7 @@ def _validate_source_ip(cls, value):
 class RequestContextV2(BaseModel):
     accountId: str
     apiId: str
-    authorizer: Optional[RequestContextV2Authorizer] = None
+    authorizer: RequestContextV2Authorizer | None = None
     domainName: str
     domainPrefix: str
     requestId: str
@@ -66,17 +66,17 @@ class APIGatewayProxyEventV2Model(BaseModel):
     routeKey: str
     rawPath: str
     rawQueryString: str
-    cookies: Optional[List[str]] = None
+    cookies: List[str] | None = None
     headers: Dict[str, str]
-    queryStringParameters: Optional[Dict[str, str]] = None
-    pathParameters: Optional[Dict[str, str]] = None
-    stageVariables: Optional[Dict[str, str]] = None
+    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
-    isBase64Encoded: Optional[bool] = None
+    body: Union[str, Type[BaseModel]] | None = None
+    isBase64Encoded: bool | None = None
 
 
 class ApiGatewayAuthorizerRequestV2(APIGatewayProxyEventV2Model):
     type: Literal["REQUEST"]
     routeArn: str
-    identitySource: Optional[List[str]] = None
+    identitySource: List[str] | None = None

aws_lambda_powertools/utilities/parser/models/appsync.py

@@ -1,16 +1,16 @@
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Union
 
 from pydantic import BaseModel, Field
 
 
 class AppSyncIamIdentity(BaseModel):
     accountId: str = Field(description="The AWS account ID of the caller.", examples=["123456789012"])
-    cognitoIdentityPoolId: Optional[str] = Field(
+    cognitoIdentityPoolId: str | None = Field(
         default=None,
         description="The Amazon Cognito identity pool ID associated with the caller.",
         examples=["us-east-1:12345678-1234-1234-1234-123456789012"],
     )
-    cognitoIdentityId: Optional[str] = Field(
+    cognitoIdentityId: str | None = Field(
         default=None,
         description="The Amazon Cognito identity ID of the caller.",
         examples=["us-east-1:12345678-1234-1234-1234-123456789012"],
@@ -29,12 +29,12 @@ class AppSyncIamIdentity(BaseModel):
         description="The Amazon Resource Name (ARN) of the IAM user.",
         examples=["arn:aws:iam::123456789012:user/appsync", "arn:aws:iam::123456789012:user/service-user"],
     )
-    cognitoIdentityAuthType: Optional[str] = Field(
+    cognitoIdentityAuthType: str | None = Field(
         default=None,
         description="Either authenticated or unauthenticated based on the identity type.",
         examples=["authenticated", "unauthenticated"],
     )
-    cognitoIdentityAuthProvider: Optional[str] = Field(
+    cognitoIdentityAuthProvider: str | None = Field(
         default=None,
         description=(
             "A comma-separated list of external identity provider information "
@@ -74,7 +74,7 @@ class AppSyncCognitoIdentity(BaseModel):
         description="The default authorization strategy for this caller (ALLOW or DENY).",
         examples=["ALLOW", "DENY"],
     )
-    groups: Optional[List[str]] = Field(
+    groups: List[str] | None = Field(
         default=None,
         description="The Cognito User Pool groups that the user belongs to.",
         examples=[["admin", "users"], ["developers"]],
@@ -115,7 +115,7 @@ class AppSyncLambdaIdentity(BaseModel):
 
 
 class AppSyncRequestModel(BaseModel):
-    domainName: Optional[str] = Field(
+    domainName: str | None = Field(
         default=None,
         description=(
             "The custom domain name used to access the GraphQL endpoint. "
@@ -190,11 +190,11 @@ class AppSyncResolverEventModel(BaseModel):
             {"page": 2, "size": 1, "name": "value"},
         ],
     )
-    identity: Optional[AppSyncIdentity] = Field(
+    identity: AppSyncIdentity | None = Field(
         default=None,
         description="Information about the caller identity (authenticated user or API key).",
     )
-    source: Optional[Dict[str, Any]] = Field(
+    source: Dict[str, Any] | None = Field(
         default=None,
         description="The parent object for the field. For top-level fields, this will be null.",
         examples=[
@@ -208,7 +208,7 @@ class AppSyncResolverEventModel(BaseModel):
     info: AppSyncInfoModel = Field(
         description="Information about the GraphQL request including selection set and field details.",
     )
-    prev: Optional[AppSyncPrevModel] = Field(
+    prev: AppSyncPrevModel | None = Field(
         default=None,
         description="Results from the previous resolver in a pipeline resolver.",
     )

aws_lambda_powertools/utilities/parser/models/appsync_events.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List, Literal, Optional
+from typing import Any, Dict, List, Literal
 
 from pydantic import BaseModel, Field
 
@@ -51,33 +51,33 @@ class AppSyncEventsEventModel(BaseModel):
 
 
 class AppSyncEventsModel(BaseModel):
-    identity: Optional[AppSyncIdentity] = Field(
+    identity: AppSyncIdentity | None = Field(
         default=None,
         description="Information about the caller identity (authenticated user or API key).",
     )
     request: AppSyncRequestModel = Field(description="Information about the GraphQL request context.")
     info: AppSyncEventsInfoModel = Field(
         description="Information about the AppSync Events operation including channel details.",
     )
-    prev: Optional[str] = Field(
+    prev: str | None = Field(
         default=None,
         description="Results from the previous operation in a pipeline resolver.",
         examples=["previous-result-data"],
     )
-    outErrors: Optional[List[str]] = Field(
+    outErrors: List[str] | None = Field(
         default=None,
         description="List of output errors that occurred during event processing.",
         examples=[["Error message 1", "Error message 2"]],
     )
-    stash: Optional[Dict[str, Any]] = Field(
+    stash: Dict[str, Any] | None = Field(
         default=None,
         description=(
             "The stash is a map that is made available inside each resolver and function mapping template. "
             "The same stash instance lives through a single resolver execution."
         ),
         examples=[{"customData": "value", "userId": "123"}],
     )
-    events: Optional[List[AppSyncEventsEventModel]] = Field(
+    events: List[AppSyncEventsEventModel] | None = Field(
         default=None,
         description="List of events being published or subscribed to in the AppSync Events operation.",
         examples=[