diff --git a/aws_lambda_powertools/utilities/parser/envelopes/apigw.py b/aws_lambda_powertools/utilities/parser/envelopes/apigw.py index 1a81124cf09..2a7b0c75bd0 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/apigw.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/apigw.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class ApiGatewayEnvelope(BaseEnvelope): """API Gateway envelope to extract data within body key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Any + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Api Gateway model {APIGatewayProxyEventModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/apigw_websocket.py b/aws_lambda_powertools/utilities/parser/envelopes/apigw_websocket.py index 37d08dec180..3f5adcde040 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/apigw_websocket.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/apigw_websocket.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import APIGatewayWebSocketMessageEventModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -16,19 +16,19 @@ class ApiGatewayWebSocketEnvelope(BaseEnvelope): """API Gateway WebSockets envelope to extract data within body key of messages routes (not disconnect or connect)""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Any + T | None Parsed detail payload with model provided """ logger.debug( diff --git a/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py b/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py index cb0c6b980d1..760f9aad15e 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import APIGatewayProxyEventV2Model if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class ApiGatewayV2Envelope(BaseEnvelope): """API Gateway V2 envelope to extract data within body key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Any + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Api Gateway model V2 {APIGatewayProxyEventV2Model}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/base.py b/aws_lambda_powertools/utilities/parser/envelopes/base.py index dbd76eafe7d..83209422e55 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/base.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/base.py @@ -44,7 +44,11 @@ def _parse(data: dict[str, Any] | Any | None, model: type[T]) -> T | None: return _parse_and_validate_event(data=data, adapter=adapter) @abstractmethod - def parse(self, data: dict[str, Any] | Any | None, model: type[T]): + def parse( + self, + data: dict[str, Any] | Any | None, + model: type[T], + ) -> T | list[T | None] | list[dict[str, T | None]] | None: """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 392c17cc425..61745f34edd 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/bedrock_agent.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import BedrockAgentEventModel, BedrockAgentFunctionEventModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class BedrockAgentEnvelope(BaseEnvelope): """Bedrock Agent envelope to extract data within input_text key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Model | None + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Bedrock Agent model {BedrockAgentEventModel}") @@ -39,19 +39,19 @@ def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model class BedrockAgentFunctionEnvelope(BaseEnvelope): """Bedrock Agent Function envelope to extract data within input_text key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Model | None + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Bedrock Agent Function model {BedrockAgentFunctionEventModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py b/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py index 0cfe151b789..a6dac2c5859 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import CloudWatchLogsModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -22,19 +22,19 @@ class CloudWatchLogsEnvelope(BaseEnvelope): Note: The record will be parsed the same way so if model is str """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 a7d56abdb11..4458a1553ea 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -19,19 +19,19 @@ class DynamoDBStreamEnvelope(BaseEnvelope): length of the list is the record's amount in the original event. """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[dict[str, Model | None]]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[dict[str, T | None]]: """Parses DynamoDB Stream records found in either NewImage and OldImage with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 c123319ca7d..ea972452564 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import EventBridgeModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class EventBridgeEnvelope(BaseEnvelope): """EventBridge envelope to extract data within detail key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Any + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with EventBridge model {EventBridgeModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/kafka.py b/aws_lambda_powertools/utilities/parser/envelopes/kafka.py index cba374730c6..da08f19b863 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kafka.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kafka.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import KafkaMskEventModel, KafkaSelfManagedEventModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -21,19 +21,19 @@ class KafkaEnvelope(BaseEnvelope): all items in the list will be parsed as str and npt as JSON (and vice versa) """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] List of records parsed with model provided """ event_source = cast(dict, data).get("eventSource") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py b/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py index 41527e03930..0085dade352 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kinesis.py @@ -8,7 +8,7 @@ from aws_lambda_powertools.utilities.parser.models import KinesisDataStreamModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -24,19 +24,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: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 e816ac877e9..d478421633d 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/kinesis_firehose.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import KinesisFirehoseModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -25,19 +25,19 @@ class KinesisFirehoseEnvelope(BaseEnvelope): https://docs.aws.amazon.com/lambda/latest/dg/services-kinesisfirehose.html """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 123cfd514b7..80aeb24930c 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/lambda_function_url.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import LambdaFunctionUrlModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class LambdaFunctionUrlEnvelope(BaseEnvelope): """Lambda function URL envelope to extract data within body key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Any + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with Lambda function URL model {LambdaFunctionUrlModel}") diff --git a/aws_lambda_powertools/utilities/parser/envelopes/sns.py b/aws_lambda_powertools/utilities/parser/envelopes/sns.py index 98e198c898d..c6d21231e60 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/sns.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/sns.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import SnsModel, SnsNotificationModel, SqsModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -22,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: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] List of records parsed with model provided """ logger.debug(f"Parsing incoming data with SNS model {SnsModel}") @@ -54,19 +54,19 @@ class SnsSqsEnvelope(BaseEnvelope): 3. Finally, parse provided model against payload extracted """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 9c64808d3ca..9fe42aed4da 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/sqs.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/sqs.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import SqsModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -22,19 +22,19 @@ class SqsEnvelope(BaseEnvelope): all items in the list will be parsed as str and not as JSON (and vice versa) """ - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> list[Model | None]: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | None]: """Parses records found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - list + list[T | None] 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 42facf8d279..38c454b9bd8 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/vpc_lattice.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import VpcLatticeModel if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class VpcLatticeEnvelope(BaseEnvelope): """Amazon VPC Lattice envelope to extract data within body key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Model | None + T | 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 d70a68296a0..f1a218b46a4 100644 --- a/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py +++ b/aws_lambda_powertools/utilities/parser/envelopes/vpc_latticev2.py @@ -7,7 +7,7 @@ from aws_lambda_powertools.utilities.parser.models import VpcLatticeV2Model if TYPE_CHECKING: - from aws_lambda_powertools.utilities.parser.types import Model + from aws_lambda_powertools.utilities.parser.types import T logger = logging.getLogger(__name__) @@ -15,19 +15,19 @@ class VpcLatticeV2Envelope(BaseEnvelope): """Amazon VPC Lattice envelope to extract data within body key""" - def parse(self, data: dict[str, Any] | Any | None, model: type[Model]) -> Model | None: + def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> T | None: """Parses data found with model provided Parameters ---------- data : dict Lambda event to be parsed - model : type[Model] + model : type[T] Data model provided to parse after extracting data using envelope Returns ------- - Model | None + T | None Parsed detail payload with model provided """ logger.debug(f"Parsing incoming data with VPC Lattice V2 model {VpcLatticeV2Model}") diff --git a/aws_lambda_powertools/utilities/parser/functions.py b/aws_lambda_powertools/utilities/parser/functions.py index 72dde64f12f..d2acb9a7965 100644 --- a/aws_lambda_powertools/utilities/parser/functions.py +++ b/aws_lambda_powertools/utilities/parser/functions.py @@ -16,7 +16,7 @@ logger = logging.getLogger(__name__) -def _retrieve_or_set_model_from_cache(model: type[T]) -> TypeAdapter: +def _retrieve_or_set_model_from_cache(model: type[T]) -> TypeAdapter[T]: """ Retrieves or sets a TypeAdapter instance from the cache for the given model. @@ -49,7 +49,7 @@ def _retrieve_or_set_model_from_cache(model: type[T]) -> TypeAdapter: return CACHE_TYPE_ADAPTER[id_model] -def _parse_and_validate_event(data: dict[str, Any] | Any, adapter: TypeAdapter): +def _parse_and_validate_event(data: dict[str, Any] | Any, adapter: TypeAdapter[T]): """ Parse and validate the event data using the provided adapter. diff --git a/aws_lambda_powertools/utilities/parser/parser.py b/aws_lambda_powertools/utilities/parser/parser.py index 3afad192a01..652cc1ebf1e 100644 --- a/aws_lambda_powertools/utilities/parser/parser.py +++ b/aws_lambda_powertools/utilities/parser/parser.py @@ -124,7 +124,11 @@ def parse(event: dict[str, Any], model: type[T]) -> T: ... # pragma: no cover @overload -def parse(event: dict[str, Any], model: type[T], envelope: type[Envelope]) -> T: ... # pragma: no cover +def parse( # pragma: no cover + event: dict[str, Any], + model: type[T], + envelope: type[Envelope], +) -> T | list[T | None] | list[dict[str, T | None]] | None: ... def parse(event: dict[str, Any], model: type[T], envelope: type[Envelope] | None = None): diff --git a/examples/parser/src/bring_your_own_envelope.py b/examples/parser/src/bring_your_own_envelope.py index 1fb5dea0045..ae60ac58ee3 100644 --- a/examples/parser/src/bring_your_own_envelope.py +++ b/examples/parser/src/bring_your_own_envelope.py @@ -1,5 +1,7 @@ +from __future__ import annotations + import json -from typing import Any, Dict, Optional, Type, TypeVar, Union +from typing import Any, TypeVar from pydantic import BaseModel @@ -7,11 +9,11 @@ from aws_lambda_powertools.utilities.parser.models import EventBridgeModel from aws_lambda_powertools.utilities.typing import LambdaContext -Model = TypeVar("Model", bound=BaseModel) +T = TypeVar("T") class EventBridgeEnvelope(BaseEnvelope): - 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[T]) -> T | None: if data is None: return None