feat(data-classes): AppSync Resolver Event

aws_lambda_powertools/logging/correlation_paths.py

@@ -2,5 +2,6 @@
 
 API_GATEWAY_REST = "requestContext.requestId"
 API_GATEWAY_HTTP = API_GATEWAY_REST
-APPLICATION_LOAD_BALANCER = "headers.x-amzn-trace-id"
+APPSYNC_RESOLVER = 'request.headers."x-amzn-trace-id"'
+APPLICATION_LOAD_BALANCER = 'headers."x-amzn-trace-id"'
 EVENT_BRIDGE = "id"

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -1,3 +1,5 @@
+from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncResolverEvent
+
 from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .cloud_watch_logs_event import CloudWatchLogsEvent
@@ -13,6 +15,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "AppSyncResolverEvent",
     "ALBEvent",
     "CloudWatchLogsEvent",
     "ConnectContactFlowEvent",

aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py

@@ -0,0 +1,76 @@
+import datetime
+import time
+import uuid
+from typing import Any, Dict
+
+from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+def make_id():
+    return str(uuid.uuid4())
+
+
+def aws_date():
+    now = datetime.datetime.utcnow().date()
+    return now.strftime("%Y-%m-%d")
+
+
+def aws_time():
+    now = datetime.datetime.utcnow().time()
+    return now.strftime("%H:%M:%S")
+
+
+def aws_datetime():
+    now = datetime.datetime.utcnow()
+    return now.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def aws_timestamp():
+    return int(time.time())
+
+
+class AppSyncResolver:
+    def __init__(self):
+        self._resolvers: dict = {}
+
+    def resolver(
+        self,
+        type_name: str = "*",
+        field_name: str = None,
+        include_event: bool = False,
+        include_context: bool = False,
+        **kwargs,
+    ):
+        def register_resolver(func):
+            kwargs["include_event"] = include_event
+            kwargs["include_context"] = include_context
+            self._resolvers[f"{type_name}.{field_name}"] = {
+                "func": func,
+                "config": kwargs,
+            }
+            return func
+
+        return register_resolver
+
+    def resolve(self, event: dict, context: LambdaContext) -> Any:
+        event = AppSyncResolverEvent(event)
+        resolver, config = self._resolver(event.type_name, event.field_name)
+        kwargs = self._kwargs(event, context, config)
+        return resolver(**kwargs)
+
+    def _resolver(self, type_name: str, field_name: str) -> tuple:
+        full_name = f"{type_name}.{field_name}"
+        resolver = self._resolvers.get(full_name, self._resolvers.get(f"*.{field_name}"))
+        if not resolver:
+            raise ValueError(f"No resolver found for '{full_name}'")
+        return resolver["func"], resolver["config"]
+
+    @staticmethod
+    def _kwargs(event: AppSyncResolverEvent, context: LambdaContext, config: dict) -> Dict[str, Any]:
+        kwargs = {**event.arguments}
+        if config.get("include_event", False):
+            kwargs["event"] = event
+        if config.get("include_context", False):
+            kwargs["context"] = context
+        return kwargs

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -0,0 +1,232 @@
+from typing import Any, Dict, List, Optional, Union
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, get_header_value
+
+
+def get_identity_object(identity: Optional[dict]) -> Any:
+    """Get the identity object based on the best detected type"""
+    # API_KEY authorization
+    if identity is None:
+        return None
+
+    # AMAZON_COGNITO_USER_POOLS authorization
+    if "sub" in identity:
+        return AppSyncIdentityCognito(identity)
+
+    # AWS_IAM authorization
+    return AppSyncIdentityIAM(identity)
+
+
+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 cognito_identity_id(self) -> str:
+        """The Amazon Cognito identity ID of the caller."""
+        return self["cognitoIdentityId"]
+
+    @property
+    def user_arn(self) -> str:
+        """The ARN of the IAM user."""
+        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) -> List[str]:
+        """List of OIDC groups"""
+        return self["groups"]
+
+    @property
+    def issuer(self) -> str:
+        """The token issuer."""
+        return self["issuer"]
+
+
+class AppSyncResolverEventInfo(DictWrapper):
+    """The info section contains information about the GraphQL request"""
+
+    @property
+    def field_name(self) -> str:
+        """The name of the field that is currently being resolved."""
+        return self["fieldName"]
+
+    @property
+    def parent_type_name(self) -> str:
+        """The name of the parent type for the field that is currently being resolved."""
+        return self["parentTypeName"]
+
+    @property
+    def variables(self) -> Dict[str, str]:
+        """A map which holds all variables that are passed into the GraphQL request."""
+        return self.get("variables")
+
+    @property
+    def selection_set_list(self) -> List[str]:
+        """A list representation of the fields in the GraphQL selection set. Fields that are aliased will
+        only be referenced by the alias name, not the field name."""
+        return self.get("selectionSetList")
+
+    @property
+    def selection_set_graphql(self) -> Optional[str]:
+        """A string representation of the selection set, formatted as GraphQL schema definition language (SDL).
+        Although fragments are not be merged into the selection set, inline fragments are preserved."""
+        return self.get("selectionSetGraphQL")
+
+
+class AppSyncResolverEvent(DictWrapper):
+    """AppSync resolver event
+
+    **NOTE:** AppSync Resolver Events can come in various shapes this data class
+    supports both Amplify GraphQL directive @function and Direct Lambda Resolver
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/appsync/latest/devguide/resolver-context-reference.html
+    - https://docs.amplify.aws/cli/graphql-transformer/function#structure-of-the-function-event
+    """
+
+    def __init__(self, data: dict):
+        super().__init__(data)
+
+        info: dict = data.get("info")
+        if not info:
+            info = {"fieldName": self.get("fieldName"), "parentTypeName": self.get("typeName")}
+
+        self._info = AppSyncResolverEventInfo(info)
+
+    @property
+    def type_name(self) -> str:
+        """The name of the parent type for the field that is currently being resolved."""
+        return self.info.parent_type_name
+
+    @property
+    def field_name(self) -> str:
+        """The name of the field that is currently being resolved."""
+        return self.info.field_name
+
+    @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.
+
+        Depending of 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")
+
+    @property
+    def request_headers(self) -> Dict[str, str]:
+        """Request headers"""
+        return self["request"]["headers"]
+
+    @property
+    def prev_result(self) -> Optional[Dict[str, any]]:
+        """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) -> Optional[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")
+
+    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,21 @@ def get(self, key: str) -> Optional[Any]:
         return self._data.get(key)
 
 
+def get_header_value(headers: Dict[str, str], name: str, default_value: str, case_sensitive: bool) -> Optional[str]:
+    """Get header value by name"""
+    if case_sensitive:
+        return headers.get(name, default_value)
+
+    name_lower = name.lower()
+
+    return next(
+        # Iterate over the dict and do a case insensitive key comparison
+        (value for key, value in headers.items() if key.lower() == name_lower),
+        # Default value is returned if no matches was found
+        default_value,
+    )
+
+
 class BaseProxyEvent(DictWrapper):
     @property
     def headers(self) -> Dict[str, str]:
@@ -72,7 +87,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)

docs/core/logger.md

@@ -516,7 +516,7 @@ When logging exceptions, Logger will add new keys named `exception_name` and `ex
        "timestamp": "2020-08-28 18:11:38,886",
        "service": "service_undefined",
        "sampling_rate": 0.0,
-       "exception_name":"ValueError",
+       "exception_name": "ValueError",
        "exception": "Traceback (most recent call last):\n  File \"<input>\", line 2, in <module>\nValueError: something went wrong"
     }
     ```