diff --git a/aws_lambda_powertools/utilities/parser/envelopes/apigw.py b/aws_lambda_powertools/utilities/parser/envelopes/apigw.py index 2e31c74e796..1a81124cf09 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/apigw.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/apigw.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventModel -from ..models import APIGatewayProxyEventModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,14 +15,14 @@ class ApiGatewayEnvelope(BaseEnvelope): """API Gateway envelope to extract data within body key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns diff --git a/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py b/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py index c82e96ba358..cb0c6b980d1 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventV2Model -from ..models import APIGatewayProxyEventV2Model -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,14 +15,14 @@ class ApiGatewayV2Envelope(BaseEnvelope): """API Gateway V2 envelope to extract data within body key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns diff --git a/aws_lambda_powertools/utilities/parser/envelopes/base.py b/aws_lambda_powertools/utilities/parser/envelopes/base.py index eefdbb7f042..14b5c0f0a32 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/base.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/base.py @@ -2,10 +2,12 @@ import logging from abc import ABC, abstractmethod -from typing import Any, Dict, Optional, TypeVar, Union +from typing import TYPE_CHECKING, Any, TypeVar from aws_lambda_powertools.utilities.parser.functions import _retrieve_or_set_model_from_cache -from aws_lambda_powertools.utilities.parser.types import T + +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -14,12 +16,12 @@ class BaseEnvelope(ABC): """ABC implementation for creating a supported Envelope""" @staticmethod - def _parse(data: Optional[Union[Dict[str, Any], Any]], model: type[T]) -> Union[T, None]: + def _parse(data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses envelope data against model provided Parameters ---------- - data : Dict + data : dict Data to be parsed and validated model : type[T] Data model to parse and validate data against @@ -43,7 +45,7 @@ def _parse(data: Optional[Union[Dict[str, Any], Any]], model: type[T]) -> Union[ return adapter.validate_python(data) @abstractmethod - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: type[T]): + def parse(self, data: dict[str, Any] | Any | None, model: type[T]): """Implementation to parse data against envelope model, then against the data model NOTE: Call `_parse` method to fully parse data with model provided. diff --git a/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py b/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py index fa0b219fe3c..3d234999116 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import BedrockAgentEventModel -from ..models import BedrockAgentEventModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,19 +15,19 @@ class BedrockAgentEnvelope(BaseEnvelope): """Bedrock Agent envelope to extract data within input_text key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - Optional[Model] + Model | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Bedrock Agent model {BedrockAgentEventModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py b/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py index 5ff5cd2a66b..0cfe151b789 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py @@ -1,15 +1,19 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import CloudWatchLogsModel -from ..models import CloudWatchLogsModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) class CloudWatchLogsEnvelope(BaseEnvelope): - """CloudWatch Envelope to extract a List of log records. + """CloudWatch Envelope to extract a list of log records. The record's body parameter is a string (after being base64 decoded and gzipped), though it can also be a JSON encoded string. @@ -18,19 +22,19 @@ class CloudWatchLogsEnvelope(BaseEnvelope): Note: The record will be parsed the same way so if model is str """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with SNS model {CloudWatchLogsModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py b/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py index 944792ba7c0..a7d56abdb11 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamModel -from ..models import DynamoDBStreamModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -15,19 +19,19 @@ class DynamoDBStreamEnvelope(BaseEnvelope): length of the list is the record's amount in the original event. """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Dict[str, Optional[Model]]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[dict[str, Model | None]]: """Parses DynamoDB Stream records found in either NewImage and OldImage with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of dictionaries with NewImage and OldImage records parsed with model provided """ logger.debug(f"Parsing incoming data with DynamoDB Stream model {DynamoDBStreamModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py b/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py index 8476494fb6d..c123319ca7d 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import EventBridgeModel -from ..models import EventBridgeModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,14 +15,14 @@ class EventBridgeEnvelope(BaseEnvelope): """EventBridge envelope to extract data within detail key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns diff --git a/aws_lambda_powertools/utilities/parser/envelopes/kafka.py b/aws_lambda_powertools/utilities/parser/envelopes/kafka.py index dee1742e324..cba374730c6 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kafka.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kafka.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union, cast +from typing import TYPE_CHECKING, Any, cast + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import KafkaMskEventModel, KafkaSelfManagedEventModel -from ..models import KafkaMskEventModel, KafkaSelfManagedEventModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -17,23 +21,23 @@ class KafkaEnvelope(BaseEnvelope): all items in the list will be parsed as str and npt as JSON (and vice versa) """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ event_source = cast(dict, data).get("eventSource") - model_parse_event: Union[Type[KafkaMskEventModel], Type[KafkaSelfManagedEventModel]] = ( + model_parse_event: type[KafkaMskEventModel | KafkaSelfManagedEventModel] = ( KafkaMskEventModel if event_source == "aws:kafka" else KafkaSelfManagedEventModel ) diff --git a/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py b/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py index 59ea623305c..a4d484931df 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union, cast +from typing import TYPE_CHECKING, Any, cast + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import KinesisDataStreamModel -from ..models import KinesisDataStreamModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -19,19 +23,19 @@ class KinesisDataStreamEnvelope(BaseEnvelope): all items in the list will be parsed as str and not as JSON (and vice versa) """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with Kinesis model {KinesisDataStreamModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py b/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py index a7d89cb0ae5..e816ac877e9 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union, cast +from typing import TYPE_CHECKING, Any, cast + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import KinesisFirehoseModel -from ..models import KinesisFirehoseModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -21,19 +25,19 @@ class KinesisFirehoseEnvelope(BaseEnvelope): https://docs.aws.amazon.com/lambda/latest/dg/services-kinesisfirehose.html """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with Kinesis Firehose model {KinesisFirehoseModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py b/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py index 4f91d6db02c..123cfd514b7 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import LambdaFunctionUrlModel -from ..models import LambdaFunctionUrlModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,14 +15,14 @@ class LambdaFunctionUrlEnvelope(BaseEnvelope): """Lambda function URL envelope to extract data within body key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns diff --git a/aws_lambda_powertools/utilities/parser/envelopes/sns.py b/aws_lambda_powertools/utilities/parser/envelopes/sns.py index 4cd29c311dc..98e198c898d 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/sns.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/sns.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union, cast +from typing import TYPE_CHECKING, Any, cast + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import SnsModel, SnsNotificationModel, SqsModel -from ..models import SnsModel, SnsNotificationModel, SqsModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -18,19 +22,19 @@ class SnsEnvelope(BaseEnvelope): all items in the list will be parsed as str and npt as JSON (and vice versa) """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with SNS model {SnsModel}") @@ -50,19 +54,19 @@ class SnsSqsEnvelope(BaseEnvelope): 3. Finally, parse provided model against payload extracted """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with SQS model {SqsModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/sqs.py b/aws_lambda_powertools/utilities/parser/envelopes/sqs.py index d2c4504ecfc..faf9d44e559 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/sqs.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/sqs.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, List, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import SqsModel -from ..models import SqsModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -18,19 +22,19 @@ class SqsEnvelope(BaseEnvelope): all items in the list will be parsed as str and npt as JSON (and vice versa) """ - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> List[Optional[Model]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: """Parses records found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - List + list List of records parsed with model provided """ logger.debug(f"Parsing incoming data with SQS model {SqsModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py b/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py index 850b64e106b..42facf8d279 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import VpcLatticeModel -from ..models import VpcLatticeModel -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,19 +15,19 @@ class VpcLatticeEnvelope(BaseEnvelope): """Amazon VPC Lattice envelope to extract data within body key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - Optional[Model] + Model | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with VPC Lattice model {VpcLatticeModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py b/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py index ea85b20be8f..d70a68296a0 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py @@ -1,9 +1,13 @@ +from __future__ import annotations + import logging -from typing import Any, Dict, Optional, Type, Union +from typing import TYPE_CHECKING, Any + +from aws_lambda_powertools.utilities.parser.envelopes.base import BaseEnvelope +from aws_lambda_powertools.utilities.parser.models import VpcLatticeV2Model -from ..models import VpcLatticeV2Model -from ..types import Model -from .base import BaseEnvelope +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parser.types import Model logger = logging.getLogger(__name__) @@ -11,19 +15,19 @@ class VpcLatticeV2Envelope(BaseEnvelope): """Amazon VPC Lattice envelope to extract data within body key""" - def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model]) -> Optional[Model]: + def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: """Parses data found with model provided Parameters ---------- - data : Dict + data : dict Lambda event to be parsed - model : Type[Model] + model : type[Model] Data model provided to parse after extracting data using envelope Returns ------- - Optional[Model] + Model | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with VPC Lattice V2 model {VpcLatticeV2Model}") diff --git a/tests/e2e/parser/handlers/handler_with_union_tag.py b/tests/e2e/parser/handlers/handler_with_union_tag.py index d822dd99a27..9dbc81662b4 100644 --- a/tests/e2e/parser/handlers/handler_with_union_tag.py +++ b/tests/e2e/parser/handlers/handler_with_union_tag.py @@ -1,8 +1,9 @@ from __future__ import annotations -from typing import Annotated, Literal, Union +from typing import Literal from pydantic import BaseModel, Field +from typing_extensions import Annotated from aws_lambda_powertools.utilities.parser import event_parser from aws_lambda_powertools.utilities.typing import LambdaContext @@ -24,7 +25,7 @@ class PartialFailureCallback(BaseModel): error_msg: str -OrderCallback = Annotated[Union[SuccessCallback, ErrorCallback, PartialFailureCallback], Field(discriminator="status")] +OrderCallback = Annotated[SuccessCallback | ErrorCallback | PartialFailureCallback, Field(discriminator="status")] @event_parser