feat(data-classes): Add AppSync resolver utilities

Changes:
* Add helper functions to generate GraphQL scalar types
* AppSyncResolver decorator which works with AppSyncResolverEvent

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/appsync_resolver_utils.py

@@ -0,0 +1,136 @@
+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():
+    """A unique identifier for an object. This scalar is serialized like a String but isn't meant to be
+    human-readable."""
+    return str(uuid.uuid4())
+
+
+def aws_date():
+    """AWSDate - An extended ISO 8601 date string in the format YYYY-MM-DD"""
+    now = datetime.datetime.utcnow().date()
+    return now.strftime("%Y-%m-%d")
+
+
+def aws_time():
+    """AWSTime - An extended ISO 8601 time string in the format hh:mm:ss.sss"""
+    now = datetime.datetime.utcnow().time()
+    return now.strftime("%H:%M:%S")
+
+
+def aws_datetime():
+    """AWSDateTime - An extended ISO 8601 date and time string in the format YYYY-MM-DDThh:mm:ss.sssZ."""
+    now = datetime.datetime.utcnow()
+    return now.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def aws_timestamp():
+    """AWSTimestamp - An integer value representing the number of seconds before or after 1970-01-01-T00:00Z."""
+    return int(time.time())
+
+
+class AppSyncResolver:
+    """AppSync resolver decorator utility"""
+
+    def __init__(self):
+        self._resolvers: dict = {}
+
+    def resolver(self, type_name: str = "*", field_name: str = None, **kwargs):
+        """Registers the resolver for field_name
+
+        Parameters
+        ----------
+        type_name : str
+            Type name
+        field_name : str
+            Field name
+        kwargs :
+            Keyword arguments
+        """
+
+        def register_resolver(func):
+            self._resolvers[f"{type_name}.{field_name}"] = {
+                "func": func,
+                "config": kwargs,
+            }
+            return func
+
+        return register_resolver
+
+    def resolve(self, event: dict, context: LambdaContext) -> Any:
+        """Resolve field_name
+
+        Parameters
+        ----------
+        event : dict
+            Lambda event
+        context : LambdaContext
+            Lambda context
+
+        Returns
+        -------
+        Any
+            Returns the result of the resolver
+
+        Raises
+        -------
+        ValueError
+            If we could not find a field resolver
+        """
+        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:
+        """Find resolver for field_name
+
+        Parameters
+        ----------
+        type_name : str
+            Type name
+        field_name : str
+            Field name
+
+        Returns
+        -------
+        tuple
+            callable function and configuration
+        """
+        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]:
+        """Get the keyword arguments
+
+        Parameters
+        ----------
+        event : AppSyncResolverEvent
+            Lambda event
+        context : LambdaContext
+            Lambda context
+        config : dict
+            Configuration settings
+
+        Returns
+        -------
+        dict
+            Returns keyword arguments
+        """
+        kwargs = {**event.arguments}
+        if config.get("include_event", False):
+            kwargs["event"] = event
+        if config.get("include_context", False):
+            kwargs["context"] = context
+        return kwargs

tests/functional/appsync/test_appsync_resolver_utils.py

@@ -0,0 +1,109 @@
+import datetime
+import json
+import os
+
+import pytest
+
+from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
+from aws_lambda_powertools.utilities.data_classes.appsync_resolver_utils import (
+    AppSyncResolver,
+    aws_date,
+    aws_datetime,
+    aws_time,
+    aws_timestamp,
+    make_id,
+)
+
+
+def load_event(file_name: str) -> dict:
+    full_file_name = os.path.dirname(os.path.realpath(__file__)) + "/../../events/" + file_name
+    with open(full_file_name) as fp:
+        return json.load(fp)
+
+
+def test_direct_resolver():
+    _event = load_event("appSyncDirectResolver.json")
+
+    app = AppSyncResolver()
+
+    @app.resolver(field_name="createSomething", include_context=True)
+    def create_something(context, id: str):  # noqa AA03 VNE003
+        assert context == {}
+        return id
+
+    def handler(event, context):
+        return app.resolve(event, context)
+
+    result = handler(_event, {})
+    assert result == "my identifier"
+
+
+def test_amplify_resolver():
+    _event = load_event("appSyncResolverEvent.json")
+
+    app = AppSyncResolver()
+
+    @app.resolver(type_name="Merchant", field_name="locations", include_event=True)
+    def get_location(event: AppSyncResolverEvent, page: int, size: int, name: str):
+        assert event is not None
+        assert page == 2
+        assert size == 1
+        return name
+
+    def handler(event, context):
+        return app.resolve(event, context)
+
+    result = handler(_event, {})
+    assert result == "value"
+
+
+def test_resolver_no_params():
+    app = AppSyncResolver()
+
+    @app.resolver(type_name="Query", field_name="noParams")
+    def no_params():
+        return "no_params has no params"
+
+    event = {"typeName": "Query", "fieldName": "noParams", "arguments": {}}
+    result = app.resolve(event, None)
+
+    assert result == "no_params has no params"
+
+
+def test_resolver_value_error():
+    app = AppSyncResolver()
+
+    with pytest.raises(ValueError) as exp:
+        event = {"typeName": "type", "fieldName": "field", "arguments": {}}
+        app.resolve(event, None)
+
+    assert exp.value.args[0] == "No resolver found for 'type.field'"
+
+
+def test_make_id():
+    uuid: str = make_id()
+    assert isinstance(uuid, str)
+    assert len(uuid) == 36
+
+
+def test_aws_date():
+    date_str = aws_date()
+    assert isinstance(date_str, str)
+    assert datetime.datetime.strptime(date_str, "%Y-%m-%d")
+
+
+def test_aws_time():
+    time_str = aws_time()
+    assert isinstance(time_str, str)
+    assert datetime.datetime.strptime(time_str, "%H:%M:%S")
+
+
+def test_aws_datetime():
+    datetime_str = aws_datetime()
+    assert isinstance(datetime_str, str)
+    assert datetime.datetime.strptime(datetime_str, "%Y-%m-%dT%H:%M:%SZ")
+
+
+def test_aws_timestamp():
+    timestamp = aws_timestamp()
+    assert isinstance(timestamp, int)