Migrating functions from v1 to v2

Author: leandrodamascena

aws_lambda_powertools/event_handler/openapi/encoders.py

@@ -10,8 +10,8 @@
 from uuid import UUID
 
 from pydantic import BaseModel
-from pydantic.color import Color
 from pydantic.types import SecretBytes, SecretStr
+from pydantic_extra_types.color import Color
 
 from aws_lambda_powertools.event_handler.openapi.compat import _model_dump
 from aws_lambda_powertools.event_handler.openapi.types import IncEx

aws_lambda_powertools/utilities/parser/models/s3.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import List, Optional
 
-from pydantic import BaseModel, root_validator
+from pydantic import BaseModel, model_validator
 from pydantic.fields import Field
 from pydantic.networks import IPvAnyNetwork
 from pydantic.types import NonNegativeFloat
@@ -101,13 +101,12 @@ class S3RecordModel(BaseModel):
     s3: S3Message
     glacierEventData: Optional[S3EventRecordGlacierEventData] = None
 
-    @root_validator(allow_reuse=True, skip_on_failure=True)  # review
+    @model_validator(mode="before")
     def validate_s3_object(cls, values):
         event_name = values.get("eventName")
-        s3_object = values.get("s3").object
-        if "ObjectRemoved" not in event_name:
-            if s3_object.size is None or s3_object.eTag is None:
-                raise ValueError("S3Object.size and S3Object.eTag are required for non-ObjectRemoved events")
+        s3_object = values.get("s3").get("object")
+        if "ObjectRemoved" not in event_name and (s3_object.get("size") is None or s3_object.get("eTag") is None):
+            raise ValueError("S3Object.size and S3Object.eTag are required for non-ObjectRemoved events")
         return values
 
 

aws_lambda_powertools/utilities/parser/models/s3_batch_operation.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, List, Optional
 
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, model_validator
 
 from aws_lambda_powertools.utilities.parser.types import Literal
 
@@ -12,14 +12,12 @@ class S3BatchOperationTaskModel(BaseModel):
     s3BucketArn: Optional[str] = None
     s3Bucket: Optional[str] = None
 
-    @validator("s3Bucket", pre=True, always=True)  # review
-    def validate_bucket(cls, current_value, values):
-        # Get the s3 bucket, either from 's3Bucket' property (invocationSchemaVersion '2.0')
-        # or from 's3BucketArn' (invocationSchemaVersion '1.0')
-        if values.get("s3BucketArn") and not current_value:
-            # Replace s3Bucket value with the value from s3BucketArn
-            return values["s3BucketArn"].split(":::")[-1]
-        return current_value
+    @model_validator(mode="before")
+    def validate_s3bucket(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        if values.get("s3BucketArn") and not values.get("s3Bucket"):
+            values["s3Bucket"] = values["s3BucketArn"].split(":::")[-1]
+
+        return values
 
 
 class S3BatchOperationJobModel(BaseModel):