Adding data class

Author: leandrodamascena

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -6,6 +6,7 @@
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .api_gateway_websocket_event import APIGatewayWebSocketEvent
 from .appsync_resolver_event import AppSyncResolverEvent
+from .appsync_resolver_events_event import AppSyncResolverEventsEvent
 from .aws_config_rule_event import AWSConfigRuleEvent
 from .bedrock_agent_event import BedrockAgentEvent
 from .cloud_watch_alarm_event import (
@@ -55,6 +56,7 @@
     "APIGatewayWebSocketEvent",
     "SecretsManagerEvent",
     "AppSyncResolverEvent",
+    "AppSyncResolverEventsEvent",
     "ALBEvent",
     "BedrockAgentEvent",
     "CloudWatchAlarmData",

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -26,6 +26,50 @@ def get_identity_object(identity: dict | None) -> Any:
     return AppSyncIdentityIAM(identity)
 
 
+class AppSyncEventBase(DictWrapper):
+    """AppSync resolver event base to work with AppSync GraphQL + Events"""
+
+    @property
+    def request_headers(self) -> dict[str, str]:
+        """Request headers"""
+        return CaseInsensitiveDict(self["request"]["headers"])
+
+    @property
+    def domain_name(self) -> str | None:
+        """The domain name when using custom domain"""
+        return self["request"].get("domainName")
+
+    @property
+    def prev_result(self) -> dict[str, Any] | None:
+        """It represents the result of whatever previous operation was executed in a pipeline resolver."""
+        prev = self.get("prev")
+        if not prev:
+            return None
+        return prev.get("result")
+
+    @property
+    def stash(self) -> dict:
+        """The stash is a map that is made available inside each resolver and function mapping template.
+        The same stash instance lives through a single resolver execution. This means that you can use the
+        stash to pass arbitrary data across request and response mapping templates, and across functions in
+        a pipeline resolver."""
+        return self.get("stash") or {}
+
+    @property
+    def identity(self) -> AppSyncIdentityIAM | AppSyncIdentityCognito | None:
+        """An object that contains information about the caller.
+
+        Depending on the type of identify found:
+
+        - API_KEY authorization - returns None
+        - AWS_IAM authorization - returns AppSyncIdentityIAM
+        - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
+        - AWS_LAMBDA authorization - returns None - NEED TO TEST
+        - OPENID_CONNECT authorization - returns None - NEED TO TEST
+        """
+        return get_identity_object(self.get("identity"))
+
+
 class AppSyncIdentityIAM(DictWrapper):
     """AWS_IAM authorization"""
 
@@ -141,7 +185,7 @@ def selection_set_graphql(self) -> str | None:
         return self.get("selectionSetGraphQL")
 
 
-class AppSyncResolverEvent(DictWrapper):
+class AppSyncResolverEvent(AppSyncEventBase):
     """AppSync resolver event
 
     **NOTE:** AppSync Resolver Events can come in various shapes this data class
@@ -178,49 +222,16 @@ def arguments(self) -> dict[str, Any]:
         """A map that contains all GraphQL arguments for this field."""
         return self["arguments"]
 
-    @property
-    def identity(self) -> AppSyncIdentityIAM | AppSyncIdentityCognito | None:
-        """An object that contains information about the caller.
-
-        Depending on the type of identify found:
-
-        - API_KEY authorization - returns None
-        - AWS_IAM authorization - returns AppSyncIdentityIAM
-        - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
-        """
-        return get_identity_object(self.get("identity"))
-
     @property
     def source(self) -> dict[str, Any]:
         """A map that contains the resolution of the parent field."""
         return self.get("source") or {}
 
-    @property
-    def request_headers(self) -> dict[str, str]:
-        """Request headers"""
-        return CaseInsensitiveDict(self["request"]["headers"])
-
-    @property
-    def prev_result(self) -> dict[str, Any] | None:
-        """It represents the result of whatever previous operation was executed in a pipeline resolver."""
-        prev = self.get("prev")
-        if not prev:
-            return None
-        return prev.get("result")
-
     @property
     def info(self) -> AppSyncResolverEventInfo:
         """The info section contains information about the GraphQL request."""
         return self._info
 
-    @property
-    def stash(self) -> dict:
-        """The stash is a map that is made available inside each resolver and function mapping template.
-        The same stash instance lives through a single resolver execution. This means that you can use the
-        stash to pass arbitrary data across request and response mapping templates, and across functions in
-        a pipeline resolver."""
-        return self.get("stash") or {}
-
     @overload
     def get_header_value(
         self,

aws_lambda_powertools/utilities/data_classes/appsync_resolver_events_event.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from typing import Any
+
+from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncEventBase
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class AppSyncResolverEventsInfo(DictWrapper):
+
+    @property
+    def channel(self) -> dict[str, Any]:
+        """Channel details including path and segments"""
+        return self["channel"]
+
+    @property
+    def channel_path(self) -> str:
+        """Provides direct access to the 'path' attribute within the 'channel' object."""
+        return self["channel"]["path"]
+
+    @property
+    def channel_segments(self) -> list[str]:
+        """Provides direct access to the 'segments' attribute within the 'channel' object."""
+        return self["channel"]["segments"]
+
+    @property
+    def channel_namespace(self) -> dict:
+        """Namespace configuration for the channel"""
+        return self["channelNamespace"]
+
+    @property
+    def operation(self) -> str:
+        """The operation being performed (e.g., PUBLISH, SUBSCRIBE)"""
+        return self["operation"]
+
+
+class AppSyncResolverEventsEvent(AppSyncEventBase):
+    """AppSync resolver event events
+
+    Documentation:
+    -------------
+    - TBD
+    """
+
+    @property
+    def events(self) -> list[dict[str, Any]]:
+        """The payload sent to Lambda"""
+        return self.get("events") or [{}]
+
+    @property
+    def out_errors(self) -> list:
+        """The outErrors property"""
+        return self.get("outErrors") or []
+
+    @property
+    def info(self) -> AppSyncResolverEventsInfo:
+        "The info containing information about channel, namespace, and event"
+        return AppSyncResolverEventsInfo(self["info"])

tests/events/appSyncEventsEvent.json

@@ -0,0 +1,70 @@
+{
+  "identity":"None",
+  "result":"None",
+  "request":{
+     "headers": {
+      "x-forwarded-for": "1.1.1.1, 2.2.2.2",
+      "cloudfront-viewer-country": "US",
+      "cloudfront-is-tablet-viewer": "false",
+      "via": "2.0 xxxxxxxxxxxxxxxx.cloudfront.net (CloudFront)",
+      "cloudfront-forwarded-proto": "https",
+      "origin": "https://us-west-1.console.aws.amazon.com",
+      "content-length": "217",
+      "accept-language": "en-US,en;q=0.9",
+      "host": "xxxxxxxxxxxxxxxx.appsync-api.us-west-1.amazonaws.com",
+      "x-forwarded-proto": "https",
+      "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.83 Safari/537.36",
+      "accept": "*/*",
+      "cloudfront-is-mobile-viewer": "false",
+      "cloudfront-is-smarttv-viewer": "false",
+      "accept-encoding": "gzip, deflate, br",
+      "referer": "https://us-west-1.console.aws.amazon.com/appsync/home?region=us-west-1",
+      "content-type": "application/json",
+      "sec-fetch-mode": "cors",
+      "x-amz-cf-id": "3aykhqlUwQeANU-HGY7E_guV5EkNeMMtwyOgiA==",
+      "x-amzn-trace-id": "Root=1-5f512f51-fac632066c5e848ae714",
+      "authorization": "eyJraWQiOiJScWFCSlJqYVJlM0hrSnBTUFpIcVRXazNOW...",
+      "sec-fetch-dest": "empty",
+      "x-amz-user-agent": "AWS-Console-AppSync/",
+      "cloudfront-is-desktop-viewer": "true",
+      "sec-fetch-site": "cross-site",
+      "x-forwarded-port": "443"
+    },
+     "domainName":"None"
+  },
+  "info":{
+     "channel":{
+        "path":"/default/channel",
+        "segments":[
+           "default",
+           "channel"
+        ]
+     },
+     "channelNamespace":{
+        "name":"default"
+     },
+     "operation":"PUBLISH"
+  },
+  "error":"None",
+  "prev":"None",
+  "stash":{
+
+  },
+  "outErrors":[
+
+  ],
+  "events":[
+     {
+        "payload":{
+           "event_1":"data_1"
+        },
+        "id":"96bc1d46-5b01-40ae-9859-173a5bb3a46a"
+     },
+     {
+        "payload":{
+           "event_2":"data_2"
+        },
+        "id":"48cde766-afe9-4c2c-9708-2fed930efe56"
+     }
+  ]
+}

tests/unit/data_classes/required_dependencies/test_appsync_events_event.py

@@ -0,0 +1,16 @@
+from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEventsEvent
+from tests.functional.utils import load_event
+
+
+def test_appsync_resolver_event():
+    raw_event = load_event("appSyncEventsEvent.json")
+    parsed_event = AppSyncResolverEventsEvent(raw_event)
+
+    assert parsed_event.events == raw_event["events"]
+    assert parsed_event.out_errors == raw_event["outErrors"]
+    assert parsed_event.domain_name == raw_event["request"]["domainName"]
+    assert parsed_event.info.channel == raw_event["info"]["channel"]
+    assert parsed_event.info.channel_path == raw_event["info"]["channel"]["path"]
+    assert parsed_event.info.channel_segments == raw_event["info"]["channel"]["segments"]
+    assert parsed_event.info.channel_namespace == raw_event["info"]["channelNamespace"]
+    assert parsed_event.info.operation == raw_event["info"]["operation"]