feat(data-classes): AppSync Resolver Event

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -1,5 +1,6 @@
 from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
+from .appsync_resolver_event import AppSyncResolverEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
@@ -13,6 +14,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "AppSyncResolverEvent",
     "ALBEvent",
     "CloudWatchLogsEvent",
     "ConnectContactFlowEvent",

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -0,0 +1,157 @@
+from typing import Dict, List, Optional, Union
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, get_header_value
+
+
+class AppSyncIdentityIAM(DictWrapper):
+    """AWS_IAM authorization"""
+
+    @property
+    def source_ip(self) -> List[str]:
+        """The source IP address of the caller received by AWS AppSync. """
+        return self["sourceIp"]
+
+    @property
+    def username(self) -> str:
+        """The user name of the authenticated user. IAM user principal"""
+        return self["username"]
+
+    @property
+    def account_id(self) -> str:
+        """The AWS account ID of the caller."""
+        return self["accountId"]
+
+    @property
+    def cognito_identity_pool_id(self) -> str:
+        """The Amazon Cognito identity pool ID associated with the caller."""
+        return self["cognitoIdentityPoolId"]
+
+    @property
+    def user_arn(self) -> str:
+        return self["userArn"]
+
+    @property
+    def cognito_identity_auth_type(self) -> str:
+        """Either authenticated or unauthenticated based on the identity type."""
+        return self["cognitoIdentityAuthType"]
+
+    @property
+    def cognito_identity_auth_provider(self) -> str:
+        """A comma separated list of external identity provider information used in obtaining the
+        credentials used to sign the request."""
+        return self["cognitoIdentityAuthProvider"]
+
+
+class AppSyncIdentityCognito(DictWrapper):
+    """AMAZON_COGNITO_USER_POOLS authorization"""
+
+    @property
+    def source_ip(self) -> List[str]:
+        """The source IP address of the caller received by AWS AppSync. """
+        return self["sourceIp"]
+
+    @property
+    def username(self) -> str:
+        """The user name of the authenticated user."""
+        return self["username"]
+
+    @property
+    def sub(self) -> str:
+        """The UUID of the authenticated user."""
+        return self["sub"]
+
+    @property
+    def claims(self) -> Dict[str, str]:
+        """The claims that the user has."""
+        return self["claims"]
+
+    @property
+    def default_auth_strategy(self) -> str:
+        """The default authorization strategy for this caller (ALLOW or DENY)."""
+        return self["defaultAuthStrategy"]
+
+    @property
+    def groups(self) -> any:
+        return self.get("groups")
+
+    @property
+    def issuer(self) -> str:
+        """The token issuer."""
+        return self["issuer"]
+
+
+def get_identity_object(identity_object: Optional[dict]) -> any:
+    # API_KEY authorization
+    if identity_object is None:
+        return None
+
+    # AMAZON_COGNITO_USER_POOLS authorization
+    if "sub" in identity_object:
+        return AppSyncIdentityCognito(identity_object)
+
+    # AWS_IAM authorization
+    return AppSyncIdentityIAM(identity_object)
+
+
+class AppSyncResolverEvent(DictWrapper):
+    """AppSync resolver event
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/appsync/latest/devguide/resolver-context-reference.html
+    """
+
+    @property
+    def type_name(self) -> str:
+        """The name of the parent type for the field that is currently being resolved."""
+        return self["typeName"]
+
+    @property
+    def field_name(self) -> str:
+        """The name of the field that is currently being resolved."""
+        return self["fieldName"]
+
+    @property
+    def arguments(self) -> Dict[str, any]:
+        """A map that contains all GraphQL arguments for this field."""
+        return self["arguments"]
+
+    @property
+    def identity(self) -> Union[None, AppSyncIdentityIAM, AppSyncIdentityCognito]:
+        """An object that contains information about the caller."""
+        return get_identity_object(self["identity"])
+
+    @property
+    def source(self) -> Dict[str, any]:
+        """A map that contains the resolution of the parent field."""
+        return self["source"]
+
+    @property
+    def request_headers(self) -> Dict[str, str]:
+        """Request headers"""
+        return self["request"]["headers"]
+
+    @property
+    def prev_result(self) -> Dict[str, any]:
+        """It represents the result of whatever previous operation was executed in a pipeline resolver."""
+        return self["prev"]["result"]
+
+    def get_header_value(
+        self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False
+    ) -> Optional[str]:
+        """Get header value by name
+
+        Parameters
+        ----------
+        name: str
+            Header name
+        default_value: str, optional
+            Default value if no value was found by name
+        case_sensitive: bool
+            Whether to use a case sensitive look up
+        Returns
+        -------
+        str, optional
+            Header value
+        """
+        return get_header_value(self.request_headers, name, default_value, case_sensitive)

aws_lambda_powertools/utilities/data_classes/common.py

@@ -20,6 +20,14 @@ def get(self, key: str) -> Optional[Any]:
         return self._data.get(key)
 
 
+def get_header_value(headers: dict, name: str, default_value: str, case_sensitive: bool) -> Optional[str]:
+    """Get header value by name"""
+    if case_sensitive:
+        return headers.get(name, default_value)
+
+    return next((value for key, value in headers.items() if name.lower() == key.lower()), default_value)
+
+
 class BaseProxyEvent(DictWrapper):
     @property
     def headers(self) -> Dict[str, str]:
@@ -72,7 +80,4 @@ def get_header_value(
         str, optional
             Header value
         """
-        if case_sensitive:
-            return self.headers.get(name, default_value)
-
-        return next((value for key, value in self.headers.items() if name.lower() == key.lower()), default_value)
+        return get_header_value(self.headers, name, default_value, case_sensitive)

tests/events/appSyncResolverEvent.json

@@ -0,0 +1,71 @@
+{
+  "typeName": "Merchant",
+  "fieldName": "locations",
+  "arguments": {
+    "page": 2,
+    "size": 1,
+    "name": "value"
+  },
+  "identity": {
+    "claims": {
+      "sub": "07920713-4526-4642-9c88-2953512de441",
+      "iss": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_POOL_ID",
+      "aud": "58rc9bf5kkti90ctmvioppukm9",
+      "event_id": "7f4c9383-abf6-48b7-b821-91643968b755",
+      "token_use": "id",
+      "auth_time": 1615366261,
+      "name": "Michael Brewer",
+      "exp": 1615369861,
+      "iat": 1615366261
+    },
+    "defaultAuthStrategy": "ALLOW",
+    "groups": null,
+    "issuer": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_POOL_ID",
+    "sourceIp": [
+      "11.215.2.22"
+    ],
+    "sub": "07920713-4526-4642-9c88-2953512de441",
+    "username": "mike"
+  },
+  "source": {
+    "name": "Value",
+    "nested": {
+      "name": "value",
+      "list": []
+    }
+  },
+  "request": {
+    "headers": {
+      "x-forwarded-for": "11.215.2.22, 64.44.173.11",
+      "cloudfront-viewer-country": "US",
+      "cloudfront-is-tablet-viewer": "false",
+      "via": "2.0 SOMETHING.cloudfront.net (CloudFront)",
+      "cloudfront-forwarded-proto": "https",
+      "origin": "https://console.aws.amazon.com",
+      "content-length": "156",
+      "accept-language": "en-US,en;q=0.9",
+      "host": "SOMETHING.appsync-api.us-east-1.amazonaws.com",
+      "x-forwarded-proto": "https",
+      "sec-gpc": "1",
+      "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) etc.",
+      "accept": "*/*",
+      "cloudfront-is-mobile-viewer": "false",
+      "cloudfront-is-smarttv-viewer": "false",
+      "accept-encoding": "gzip, deflate, br",
+      "referer": "https://console.aws.amazon.com/",
+      "content-type": "application/json",
+      "sec-fetch-mode": "cors",
+      "x-amz-cf-id": "Fo5VIuvP6V6anIEt62WzFDCK45mzM4yEdpt5BYxOl9OFqafd-WR0cA==",
+      "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
+      "authorization": "AUTH-HEADER",
+      "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"
+    }
+  },
+  "prev": {
+    "result": {}
+  }
+}

tests/functional/test_lambda_trigger_events.py

@@ -8,6 +8,7 @@
     ALBEvent,
     APIGatewayProxyEvent,
     APIGatewayProxyEventV2,
+    AppSyncResolverEvent,
     CloudWatchLogsEvent,
     EventBridgeEvent,
     KinesisStreamEvent,
@@ -16,6 +17,7 @@
     SNSEvent,
     SQSEvent,
 )
+from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncIdentityCognito
 from aws_lambda_powertools.utilities.data_classes.cognito_user_pool_event import (
     CreateAuthChallengeTriggerEvent,
     CustomMessageTriggerEvent,
@@ -874,3 +876,20 @@ def test_alb_event():
     assert event.multi_value_headers == event.get("multiValueHeaders")
     assert event.body == event["body"]
     assert event.is_base64_encoded == event["isBase64Encoded"]
+
+
+def test_appsync_resolver_event():
+    event = AppSyncResolverEvent(load_event("appSyncResolverEvent.json"))
+    assert event.type_name == "Merchant"
+    assert event.field_name == "locations"
+    assert event.arguments["name"] == "value"
+    assert event.identity["claims"]["token_use"] == "id"
+    assert event.source["name"] == "Value"
+    assert event.get_header_value("X-amzn-trace-id") == "Root=1-60488877-0b0c4e6727ab2a1c545babd0"
+    assert event.get_header_value("X-amzn-trace-id", case_sensitive=True) is None
+    assert event.get_header_value("missing", default_value="Foo") == "Foo"
+    assert event.prev_result == {}
+    assert isinstance(event.identity, AppSyncIdentityCognito)
+    identity: AppSyncIdentityCognito = event.identity
+    assert identity.claims is not None
+    assert identity.sub == "07920713-4526-4642-9c88-2953512de441"