Use BaseModel directly instead of TypeVar to avoid python3.6 limitations

Author: Michal Ploski

aws_lambda_powertools/utilities/parser/envelopes/kinesis.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type, Union, cast
 
 from ..models import KinesisDataStreamModel
 from ..types import Model
@@ -38,5 +38,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
         parsed_envelope: KinesisDataStreamModel = KinesisDataStreamModel.parse_obj(data)
         logger.debug(f"Parsing Kinesis records in `body` with {model}")
         return [
-            self._parse(data=record.kinesis.data.decode("utf-8"), model=model) for record in parsed_envelope.Records
+            self._parse(data=cast(bytes, record.kinesis.data).decode("utf-8"), model=model)
+            for record in parsed_envelope.Records
         ]

aws_lambda_powertools/utilities/parser/envelopes/sns.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type, Union, cast
 
 from ..models import SnsModel, SnsNotificationModel, SqsModel
 from ..types import Model
@@ -69,6 +69,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
         parsed_envelope = SqsModel.parse_obj(data)
         output = []
         for record in parsed_envelope.Records:
-            sns_notification: SnsNotificationModel = SnsNotificationModel.parse_raw(record.body)
+            sns_notification: SnsNotificationModel = SnsNotificationModel.parse_raw(cast(str, record.body))
             output.append(self._parse(data=sns_notification.Message, model=model))
         return output

aws_lambda_powertools/utilities/parser/models/alb.py

@@ -1,9 +1,7 @@
-from typing import Dict, Generic, Union
+from typing import Dict, Type, Union
 
 from pydantic import BaseModel
 
-from aws_lambda_powertools.utilities.parser.types import Model
-
 
 class AlbRequestContextData(BaseModel):
     targetGroupArn: str
@@ -13,10 +11,10 @@ class AlbRequestContext(BaseModel):
     elb: AlbRequestContextData
 
 
-class AlbModel(BaseModel, Generic[Model]):
+class AlbModel(BaseModel):
     httpMethod: str
     path: str
-    body: Union[str, Model]
+    body: Union[str, Type[BaseModel]]
     isBase64Encoded: bool
     headers: Dict[str, str]
     queryStringParameters: Dict[str, str]

aws_lambda_powertools/utilities/parser/models/apigw.py

@@ -1,10 +1,10 @@
 from datetime import datetime
-from typing import Any, Dict, Generic, List, Optional, Union
+from typing import Any, Dict, List, Optional, Type, Union
 
 from pydantic import BaseModel, root_validator
 from pydantic.networks import IPvAnyNetwork
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 
 class ApiGatewayUserCertValidity(BaseModel):
@@ -76,7 +76,7 @@ def check_message_id(cls, values):
         return values
 
 
-class APIGatewayProxyEventModel(BaseModel, Generic[Model]):
+class APIGatewayProxyEventModel(BaseModel):
     version: Optional[str]
     resource: str
     path: str
@@ -89,4 +89,4 @@ class APIGatewayProxyEventModel(BaseModel, Generic[Model]):
     pathParameters: Optional[Dict[str, str]]
     stageVariables: Optional[Dict[str, str]]
     isBase64Encoded: bool
-    body: Optional[Union[str, Model]]
+    body: Optional[Union[str, Type[BaseModel]]]

aws_lambda_powertools/utilities/parser/models/apigwv2.py

@@ -1,10 +1,10 @@
 from datetime import datetime
-from typing import Any, Dict, Generic, List, Optional, Union
+from typing import Any, Dict, List, Optional, Type, Union
 
 from pydantic import BaseModel, Field
 from pydantic.networks import IPvAnyNetwork
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 
 class RequestContextV2AuthorizerIamCognito(BaseModel):
@@ -56,7 +56,7 @@ class RequestContextV2(BaseModel):
     http: RequestContextV2Http
 
 
-class APIGatewayProxyEventV2Model(BaseModel, Generic[Model]):
+class APIGatewayProxyEventV2Model(BaseModel):
     version: str
     routeKey: str
     rawPath: str
@@ -67,5 +67,5 @@ class APIGatewayProxyEventV2Model(BaseModel, Generic[Model]):
     pathParameters: Optional[Dict[str, str]]
     stageVariables: Optional[Dict[str, str]]
     requestContext: RequestContextV2
-    body: Optional[Union[str, Model]]
+    body: Optional[Union[str, Type[BaseModel]]]
     isBase64Encoded: bool

aws_lambda_powertools/utilities/parser/models/cloudwatch.py

@@ -3,19 +3,17 @@
 import logging
 import zlib
 from datetime import datetime
-from typing import Generic, List, Union
+from typing import List, Type, Union
 
 from pydantic import BaseModel, Field, validator
 
-from aws_lambda_powertools.utilities.parser.types import Model
-
 logger = logging.getLogger(__name__)
 
 
-class CloudWatchLogsLogEvent(BaseModel, Generic[Model]):
+class CloudWatchLogsLogEvent(BaseModel):
     id: str  # noqa AA03 VNE003
     timestamp: datetime
-    message: Union[str, Model]
+    message: Union[str, Type[BaseModel]]
 
 
 class CloudWatchLogsDecode(BaseModel):

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -1,16 +1,16 @@
 from datetime import date
-from typing import Any, Dict, Generic, List, Optional, Union
+from typing import Any, Dict, List, Optional, Type, Union
 
 from pydantic import BaseModel
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 
-class DynamoDBStreamChangedRecordModel(BaseModel, Generic[Model]):
+class DynamoDBStreamChangedRecordModel(BaseModel):
     ApproximateCreationDateTime: Optional[date]
     Keys: Dict[str, Dict[str, Any]]
-    NewImage: Optional[Union[Dict[str, Any], Model]]
-    OldImage: Optional[Union[Dict[str, Any], Model]]
+    NewImage: Optional[Union[Dict[str, Any], Type[BaseModel]]]
+    OldImage: Optional[Union[Dict[str, Any], Type[BaseModel]]]
     SequenceNumber: str
     SizeBytes: int
     StreamViewType: Literal["NEW_AND_OLD_IMAGES", "KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE"]

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,12 +1,10 @@
 from datetime import datetime
-from typing import Any, Dict, Generic, List, Optional, Union
+from typing import Any, Dict, List, Optional, Type, Union
 
 from pydantic import BaseModel, Field
 
-from aws_lambda_powertools.utilities.parser.types import Model
 
-
-class EventBridgeModel(BaseModel, Generic[Model]):
+class EventBridgeModel(BaseModel):
     version: str
     id: str  # noqa: A003,VNE003
     source: str
@@ -15,5 +13,5 @@ class EventBridgeModel(BaseModel, Generic[Model]):
     region: str
     resources: List[str]
     detail_type: str = Field(None, alias="detail-type")
-    detail: Union[Dict[str, Any], Model]
+    detail: Union[Dict[str, Any], Type[BaseModel]]
     replay_name: Optional[str] = Field(None, alias="replay-name")

aws_lambda_powertools/utilities/parser/models/kinesis.py

@@ -1,20 +1,20 @@
 import base64
 import logging
 from binascii import Error as BinAsciiError
-from typing import Generic, List, Union
+from typing import List, Type, Union
 
 from pydantic import BaseModel, validator
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 logger = logging.getLogger(__name__)
 
 
-class KinesisDataStreamRecordPayload(BaseModel, Generic[Model]):
+class KinesisDataStreamRecordPayload(BaseModel):
     kinesisSchemaVersion: str
     partitionKey: str
     sequenceNumber: str
-    data: Union[bytes, Model]  # base64 encoded str is parsed into bytes
+    data: Union[bytes, Type[BaseModel]]  # base64 encoded str is parsed into bytes
     approximateArrivalTimestamp: float
 
     @validator("data", pre=True, allow_reuse=True)

aws_lambda_powertools/utilities/parser/models/s3_object_event.py

@@ -1,20 +1,18 @@
-from typing import Dict, Generic, Optional, Union
+from typing import Dict, Optional, Type, Union
 
 from pydantic import BaseModel, HttpUrl
 
-from aws_lambda_powertools.utilities.parser.types import Model
-
 
 class S3ObjectContext(BaseModel):
     inputS3Url: HttpUrl
     outputRoute: str
     outputToken: str
 
 
-class S3ObjectConfiguration(BaseModel, Generic[Model]):
+class S3ObjectConfiguration(BaseModel):
     accessPointArn: str
     supportingAccessPointArn: str
-    payload: Union[str, Model]
+    payload: Union[str, Type[BaseModel]]
 
 
 class S3ObjectUserRequest(BaseModel):

aws_lambda_powertools/utilities/parser/models/sns.py

@@ -1,24 +1,26 @@
 from datetime import datetime
-from typing import Dict, Generic, List, Optional, Union
+from typing import Dict, List, Optional
+from typing import Type as TypingType
+from typing import Union
 
 from pydantic import BaseModel, root_validator
 from pydantic.networks import HttpUrl
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 
 class SnsMsgAttributeModel(BaseModel):
     Type: str
     Value: str
 
 
-class SnsNotificationModel(BaseModel, Generic[Model]):
+class SnsNotificationModel(BaseModel):
     Subject: Optional[str]
     TopicArn: str
     UnsubscribeUrl: HttpUrl
     Type: Literal["Notification"]
     MessageAttributes: Optional[Dict[str, SnsMsgAttributeModel]]
-    Message: Union[str, Model]
+    Message: Union[str, TypingType[BaseModel]]
     MessageId: str
     SigningCertUrl: HttpUrl
     Signature: str

aws_lambda_powertools/utilities/parser/models/sqs.py

@@ -1,9 +1,9 @@
 from datetime import datetime
-from typing import Dict, Generic, List, Optional, Union
+from typing import Dict, List, Optional, Type, Union
 
 from pydantic import BaseModel
 
-from aws_lambda_powertools.utilities.parser.types import Literal, Model
+from aws_lambda_powertools.utilities.parser.types import Literal
 
 
 class SqsAttributesModel(BaseModel):
@@ -49,10 +49,10 @@ class SqsMsgAttributeModel(BaseModel):
     #     return values # noqa: E800
 
 
-class SqsRecordModel(BaseModel, Generic[Model]):
+class SqsRecordModel(BaseModel):
     messageId: str
     receiptHandle: str
-    body: Union[str, Model]
+    body: Union[str, Type[BaseModel]]
     attributes: SqsAttributesModel
     messageAttributes: Dict[str, SqsMsgAttributeModel]
     md5OfBody: str