feat(parser): add AppSync Events models (#6999)

* Add AppSync events model

* Add documentation + rename

Author: leandrodamascena

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -27,9 +27,8 @@
     RequestContextV2AuthorizerJwt,
     RequestContextV2Http,
 )
-from .appsync import (
-    AppSyncResolverEventModel,
-)
+from .appsync import AppSyncResolverEventModel
+from .appsync_events import AppSyncEventsModel
 from .bedrock_agent import (
     BedrockAgentEventModel,
     BedrockAgentFunctionEventModel,
@@ -170,6 +169,7 @@
     "AlbRequestContext",
     "AlbRequestContextData",
     "AppSyncResolverEventModel",
+    "AppSyncEventsModel",
     "DynamoDBStreamModel",
     "EventBridgeModel",
     "DynamoDBStreamChangedRecordModel",

aws_lambda_powertools/utilities/parser/models/appsync_events.py

@@ -0,0 +1,35 @@
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.utilities.parser.models.appsync import AppSyncIdentity, AppSyncRequestModel
+
+
+class AppSyncEventsInfoChannelModel(BaseModel):
+    path: str
+    segments: List[str]
+
+
+class AppSyncEventsInfoChannelNamespaceModel(BaseModel):
+    name: str
+
+
+class AppSyncEventsInfoModel(BaseModel):
+    channel: AppSyncEventsInfoChannelModel
+    channelNamespace: AppSyncEventsInfoChannelNamespaceModel
+    operation: Literal["PUBLISH", "SUBSCRIBE"]
+
+
+class AppSyncEventsEventModel(BaseModel):
+    id: str
+    payload: Dict[str, Any]
+
+
+class AppSyncEventsModel(BaseModel):
+    identity: Optional[AppSyncIdentity] = None
+    request: AppSyncRequestModel
+    info: AppSyncEventsInfoModel
+    prev: Optional[str] = None
+    outErrors: Optional[List[str]] = None
+    stash: Optional[Dict[str, Any]] = None
+    events: Optional[List[AppSyncEventsEventModel]] = None

docs/utilities/parser.md

@@ -112,6 +112,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **APIGatewayWebSocketConnectEventModel**    | Lambda Event Source payload for Amazon API Gateway WebSocket API $connect message             |
 | **APIGatewayWebSocketDisconnectEventModel** | Lambda Event Source payload for Amazon API Gateway WebSocket API $disconnect message          |
 | **AppSyncResolverEventModel**               | Lambda Event Source payload for AWS AppSync Resolver                                          |
+| **AppSyncEventsModel**                      | Lambda Event Source payload for AWS AppSync Events                                            |
 | **BedrockAgentEventModel**                  | Lambda Event Source payload for Bedrock Agents - OpenAPI-based                                |
 | **BedrockAgentFunctionEventModel**          | Lambda Event Source payload for Bedrock Agents - Function-based                               |
 | **CloudFormationCustomResourceCreateModel** | Lambda Event Source payload for AWS CloudFormation `CREATE` operation                         |

tests/events/appSyncEventsEvent.json

@@ -1,5 +1,5 @@
 {
-    "identity":"None",
+    "identity": null,
     "result":"None",
     "request":{
        "headers": {

tests/unit/parser/_pydantic/test_appsync_events.py

@@ -0,0 +1,24 @@
+import pytest
+
+from aws_lambda_powertools.utilities.parser import ValidationError, parse
+from aws_lambda_powertools.utilities.parser.models import AppSyncEventsModel
+from tests.functional.utils import load_event
+
+
+def test_appsync_event_model_parses_successfully():
+    """
+    Validate that a valid AppSync resolver events is correctly parsed by the model.
+    """
+    event = load_event("appSyncEventsEvent.json")
+    parsed_event = parse(event=event, model=AppSyncEventsModel)
+
+    assert isinstance(parsed_event, AppSyncEventsModel)
+
+
+def test_appsync_event_model_invalid_payload_raises():
+    """
+    Validate that parsing an invalid AppSync resolver events payload raises a ValidationError.
+    """
+    invalid_event = {"invalid": "event"}
+    with pytest.raises(ValidationError):
+        parse(event=invalid_event, model=AppSyncEventsModel)