fix(parser): fix EventBridgeModel when working with scheduled events (#6134)

* Fixing scheduler detail field

* Fixing scheduler detail field

Author: leandrodamascena

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import List, Optional
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from aws_lambda_powertools.utilities.parser.types import RawDictOrModel
 
@@ -19,3 +19,9 @@ class EventBridgeModel(BaseModel):
     detail_type: str = Field(..., alias="detail-type")
     detail: RawDictOrModel
     replay_name: Optional[str] = Field(None, alias="replay-name")
+
+    @field_validator("detail", mode="before")
+    def validate_detail(cls, v, fields):
+        # EventBridge Scheduler sends detail field as '{}' string when no payload is present
+        # See: https://github.com/aws-powertools/powertools-lambda-python/issues/6112
+        return {} if fields.data.get("source") == "aws.scheduler" and v == "{}" else v

tests/events/eventBridgeSchedulerEvent.json

@@ -0,0 +1,13 @@
+{
+  "version":"0",
+  "id":"d167b752-343a-4b28-afd6-d4de056319e8",
+  "detail-type":"Scheduled Event",
+  "source":"aws.scheduler",
+  "account":"123456789012",
+  "time":"2025-02-20T16:03:00Z",
+  "region":"us-east-1",
+  "resources":[
+     "arn:aws:scheduler:us-east-1:123456789012:schedule/default/aaaaa"
+  ],
+  "detail":"{}"
+}

tests/unit/parser/_pydantic/test_eventbridge.py

@@ -1,6 +1,7 @@
 import pytest
 
 from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, parse
+from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
 from tests.functional.utils import load_event
 from tests.unit.parser._pydantic.schemas import (
     MyAdvancedEventbridgeBusiness,
@@ -51,3 +52,10 @@ def test_handle_invalid_event_with_eventbridge_envelope():
     empty_event = {}
     with pytest.raises(ValidationError):
         parse(event=empty_event, model=MyEventbridgeBusiness, envelope=envelopes.EventBridgeEnvelope)
+
+
+def test_handle_eventbridge_scheduler():
+    raw_event = load_event("eventBridgeSchedulerEvent.json")
+    parsed_event: EventBridgeModel = EventBridgeModel(**raw_event)
+
+    assert parsed_event.detail == {}