Add AppSync events model

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 AppSyncResolverEventsModel
 from .bedrock_agent import (
     BedrockAgentEventModel,
     BedrockAgentFunctionEventModel,
@@ -170,6 +169,7 @@
     "AlbRequestContext",
     "AlbRequestContextData",
     "AppSyncResolverEventModel",
+    "AppSyncResolverEventsModel",
     "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 AppSyncResolverEventsModel(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

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 AppSyncResolverEventsModel
+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=AppSyncResolverEventsModel)
+
+    assert isinstance(parsed_event, AppSyncResolverEventsModel)
+
+
+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=AppSyncResolverEventsModel)