feat(event-handler): add appsync batch resolvers (#1998)

* update changelog with latest changes

* Add batch processing to appsync handler

* Extend router to accept List of events. Add functional test

* Add e2e tests

* Add required package

* Fix linter checks

* Refactor appsync resolver

* Refactor code to use composition instead of inheritence

* Refactor appsync event handler

* fix style

* Add support for async batch processing

* Fixing sonarcloud error

* Adding docstring + increasing coverage

* Adding missing test + increasing coverage

* Start writing docs

* Refactoring examples

* Refactoring code + examples + documentation

* Moving e2e tests to the right folder

* Moving e2e tests to the right folder

* Adding partial failure

* Adding partial failure

* Fixing docstring and examples

* Adding documentation about Handling Exceptions

* Fixing docstring

* Adding fine grained control when handling exceptions

* Adding fine grained control when handling exceptions

* docs: add intro diagram

* docs: fix wording (Tech debt)

* refactor: use async_ prefix for async code

* refactor: move router to a separate file to ease maintenance

* refactor: rename BasePublic to BaseRouter

* refactor: undo router context composition to reduce complexity and call stacks (getters)

* refactor: reduce abstractions, use explicit methods over assignments

Signed-off-by: heitorlessa <[email protected]>

* refactor: move registry to a separate file; make it private

* refactor: expand inline if for readability

Signed-off-by: heitorlessa <[email protected]>

* refactor: short circuit upfront, complex after

Renamed long var name to `event` to make it quicker to glance through.

Simplified comments by removing redundant ones.

* refactor: simplify arg name

* refactor: add debug statements

* fix(docs): use .context instead of previous ._router.context

* refactor: use kwargs for explicitness

* refactor: use return_exceptions=True to reduce call stack

Signed-off-by: heitorlessa <[email protected]>

* chore: add notes on the beauty of return_exceptions

* refactor: append  suffix in exceptions

* chore: if over elif in short-circuit

* chore: improve logging; glad I learned this new f-string trick

* chore: fix debug statement location due to null resolvers

* revert: debug graceful error flag due to non-determinism async

Signed-off-by: heitorlessa <[email protected]>

* revert: debug stmt due to mypy; moving elsewhere

Signed-off-by: heitorlessa <[email protected]>

* docs: docstring resolver (tech debt)

Signed-off-by: heitorlessa <[email protected]>

* docs: minimal batch_resolver docstring

Signed-off-by: heitorlessa <[email protected]>

* chore: complete resolver docstring

Signed-off-by: heitorlessa <[email protected]>

* Removing payload exception

* Updating poetry

* Merging from develop

* Addressing Heitor's feedback

* Refactoring to support aggregate events

* Refactoring examples + docs

* Addressing Heitor's feedback

* docs: add diagram to visualize n+1 problem

* docs: improve wording in lambda invoke

* docs: add diagram where n+1 problem shifts to Lambda runtime

* docs: add diagram where n+1 problem shifts to Lambda runtime w/ error handling

* docs: highlight lambda response for non-errors

* docs(setup): increase table of contents depth to 5 to help redis and other subsections

* Adding examples

* docs: explain N+1 problem and organize content into sub-sections

* docs: clean up batch resolvers section; add typing

* docs: clean up no-aggregate processing section

* docs: clean up raise on error section

* Adding examples

* docs: clean up async section

* docs: fix highlights, add missing code annotation

* docs: rename snippets to match advanced section

* Merging from develop

* Tests

---------

Signed-off-by: Leandro Damascena <[email protected]>
Signed-off-by: heitorlessa <[email protected]>
Signed-off-by: Heitor Lessa <[email protected]>
Co-authored-by: Release bot <[email protected]>
Co-authored-by: Michal Ploski <[email protected]>
Co-authored-by: Leandro Damascena <[email protected]>
Co-authored-by: Heitor Lessa <[email protected]>
Co-authored-by: Leandro Damascena <[email protected]>
Co-authored-by: Heitor Lessa <[email protected]>
Co-authored-by: Cavalcante Damascena <[email protected]>

Author: 6 people

Diff for: aws_lambda_powertools/event_handler/appsync.py

@@ -0,0 +1,76 @@
+import logging
+from typing import Any, Callable, Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class ResolverRegistry:
+    def __init__(self):
+        self.resolvers: Dict[str, Dict[str, Any]] = {}
+
+    def register(
+        self,
+        type_name: str = "*",
+        field_name: Optional[str] = None,
+        raise_on_error: bool = False,
+        aggregate: bool = True,
+    ) -> Callable:
+        """Registers the resolver for field_name
+
+        Parameters
+        ----------
+        type_name : str
+            Type name
+        field_name : str
+            Field name
+        raise_on_error: bool
+            A flag indicating whether to raise an error when processing batches
+            with failed items. Defaults to False, which means errors are handled without raising exceptions.
+        aggregate: bool
+            A flag indicating whether the batch items should be processed at once or individually.
+            If True (default), the batch resolver will process all items in the batch as a single event.
+            If False, the batch resolver will process each item in the batch individually.
+
+        Return
+        ----------
+        Dict
+            A dictionary with the resolver and if raise exception on error
+        """
+
+        def _register(func) -> Callable:
+            logger.debug(f"Adding resolver `{func.__name__}` for field `{type_name}.{field_name}`")
+            self.resolvers[f"{type_name}.{field_name}"] = {
+                "func": func,
+                "raise_on_error": raise_on_error,
+                "aggregate": aggregate,
+            }
+            return func
+
+        return _register
+
+    def find_resolver(self, type_name: str, field_name: str) -> Optional[Dict]:
+        """Find resolver based on type_name and field_name
+
+        Parameters
+        ----------
+        type_name : str
+            Type name
+        field_name : str
+            Field name
+        Return
+        ----------
+        Optional[Dict]
+            A dictionary with the resolver and if raise exception on error
+        """
+        logger.debug(f"Looking for resolver for type={type_name}, field={field_name}.")
+        return self.resolvers.get(f"{type_name}.{field_name}", self.resolvers.get(f"*.{field_name}"))
+
+    def merge(self, other_registry: "ResolverRegistry"):
+        """Update current registry with incoming registry
+
+        Parameters
+        ----------
+        other_registry : ResolverRegistry
+            Registry to merge from
+        """
+        self.resolvers.update(**other_registry.resolvers)

Diff for: aws_lambda_powertools/event_handler/graphql_appsync/__init__.py

@@ -0,0 +1,158 @@
+from abc import ABC, abstractmethod
+from typing import Callable, Optional
+
+
+class BaseRouter(ABC):
+    """Abstract base class for Router (resolvers)"""
+
+    @abstractmethod
+    def resolver(self, type_name: str = "*", field_name: Optional[str] = None) -> Callable:
+        """
+        Retrieve a resolver function for a specific type and field.
+
+        Parameters
+        -----------
+        type_name: str
+            The name of the type.
+        field_name: Optional[str]
+            The name of the field (default is None).
+
+        Examples
+        --------
+        ```python
+        from typing import Optional
+
+        from aws_lambda_powertools.event_handler import AppSyncResolver
+        from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
+        from aws_lambda_powertools.utilities.typing import LambdaContext
+
+        app = AppSyncResolver()
+
+        @app.resolver(type_name="Query", field_name="getPost")
+        def related_posts(event: AppSyncResolverEvent) -> Optional[list]:
+            return {"success": "ok"}
+
+        def lambda_handler(event, context: LambdaContext) -> dict:
+            return app.resolve(event, context)
+        ```
+
+        Returns
+        -------
+        Callable
+            The resolver function.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def batch_resolver(
+        self,
+        type_name: str = "*",
+        field_name: Optional[str] = None,
+        raise_on_error: bool = False,
+        aggregate: bool = True,
+    ) -> Callable:
+        """
+        Retrieve a batch resolver function for a specific type and field.
+
+        Parameters
+        -----------
+        type_name: str
+            The name of the type.
+        field_name: Optional[str]
+            The name of the field (default is None).
+        raise_on_error: bool
+            A flag indicating whether to raise an error when processing batches
+            with failed items. Defaults to False, which means errors are handled without raising exceptions.
+        aggregate: bool
+            A flag indicating whether the batch items should be processed at once or individually.
+            If True (default), the batch resolver will process all items in the batch as a single event.
+            If False, the batch resolver will process each item in the batch individually.
+
+        Examples
+        --------
+        ```python
+        from typing import Optional
+
+        from aws_lambda_powertools.event_handler import AppSyncResolver
+        from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
+        from aws_lambda_powertools.utilities.typing import LambdaContext
+
+        app = AppSyncResolver()
+
+        @app.batch_resolver(type_name="Query", field_name="getPost")
+        def related_posts(event: AppSyncResolverEvent, id) -> Optional[list]:
+            return {"post_id": id}
+
+        def lambda_handler(event, context: LambdaContext) -> dict:
+            return app.resolve(event, context)
+        ```
+
+        Returns
+        -------
+        Callable
+            The batch resolver function.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def async_batch_resolver(
+        self,
+        type_name: str = "*",
+        field_name: Optional[str] = None,
+        raise_on_error: bool = False,
+        aggregate: bool = True,
+    ) -> Callable:
+        """
+        Retrieve a batch resolver function for a specific type and field and runs async.
+
+        Parameters
+        -----------
+        type_name: str
+            The name of the type.
+        field_name: Optional[str]
+            The name of the field (default is None).
+        raise_on_error: bool
+            A flag indicating whether to raise an error when processing batches
+            with failed items. Defaults to False, which means errors are handled without raising exceptions.
+        aggregate: bool
+            A flag indicating whether the batch items should be processed at once or individually.
+            If True (default), the batch resolver will process all items in the batch as a single event.
+            If False, the batch resolver will process each item in the batch individually.
+
+        Examples
+        --------
+        ```python
+        from typing import Optional
+
+        from aws_lambda_powertools.event_handler import AppSyncResolver
+        from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
+        from aws_lambda_powertools.utilities.typing import LambdaContext
+
+        app = AppSyncResolver()
+
+        @app.async_batch_resolver(type_name="Query", field_name="getPost")
+        async def related_posts(event: AppSyncResolverEvent, id) -> Optional[list]:
+            return {"post_id": id}
+
+        def lambda_handler(event, context: LambdaContext) -> dict:
+            return app.resolve(event, context)
+        ```
+
+        Returns
+        -------
+        Callable
+            The batch resolver function.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def append_context(self, **additional_context) -> None:
+        """
+        Appends context information available under any route.
+
+        Parameters
+        -----------
+        **additional_context: dict
+            Additional context key-value pairs to append.
+        """
+        raise NotImplementedError

Diff for: aws_lambda_powertools/event_handler/graphql_appsync/_registry.py

@@ -0,0 +1,10 @@
+class ResolverNotFoundError(Exception):
+    """
+    When a resolver is not found during a lookup.
+    """
+
+
+class InvalidBatchResponse(Exception):
+    """
+    When a batch response something different from a List
+    """

Diff for: aws_lambda_powertools/event_handler/graphql_appsync/base.py

@@ -0,0 +1,53 @@
+from typing import Callable, Optional
+
+from aws_lambda_powertools.event_handler.graphql_appsync._registry import ResolverRegistry
+from aws_lambda_powertools.event_handler.graphql_appsync.base import BaseRouter
+
+
+class Router(BaseRouter):
+    context: dict
+
+    def __init__(self):
+        self.context = {}  # early init as customers might add context before event resolution
+        self._resolver_registry = ResolverRegistry()
+        self._batch_resolver_registry = ResolverRegistry()
+        self._async_batch_resolver_registry = ResolverRegistry()
+
+    def resolver(self, type_name: str = "*", field_name: Optional[str] = None) -> Callable:
+        return self._resolver_registry.register(field_name=field_name, type_name=type_name)
+
+    def batch_resolver(
+        self,
+        type_name: str = "*",
+        field_name: Optional[str] = None,
+        raise_on_error: bool = False,
+        aggregate: bool = True,
+    ) -> Callable:
+        return self._batch_resolver_registry.register(
+            field_name=field_name,
+            type_name=type_name,
+            raise_on_error=raise_on_error,
+            aggregate=aggregate,
+        )
+
+    def async_batch_resolver(
+        self,
+        type_name: str = "*",
+        field_name: Optional[str] = None,
+        raise_on_error: bool = False,
+        aggregate: bool = True,
+    ) -> Callable:
+        return self._async_batch_resolver_registry.register(
+            field_name=field_name,
+            type_name=type_name,
+            raise_on_error=raise_on_error,
+            aggregate=aggregate,
+        )
+
+    def append_context(self, **additional_context):
+        """Append key=value data as routing context"""
+        self.context.update(**additional_context)
+
+    def clear_context(self):
+        """Resets routing context"""
+        self.context.clear()

Diff for: aws_lambda_powertools/event_handler/graphql_appsync/exceptions.py

@@ -184,9 +184,9 @@ def identity(self) -> Union[None, AppSyncIdentityIAM, AppSyncIdentityCognito]:
         return get_identity_object(self.get("identity"))
 
     @property
-    def source(self) -> Optional[Dict[str, Any]]:
+    def source(self) -> Dict[str, Any]:
         """A map that contains the resolution of the parent field."""
-        return self.get("source")
+        return self.get("source") or {}
 
     @property
     def request_headers(self) -> Dict[str, str]:

Diff for: aws_lambda_powertools/event_handler/graphql_appsync/router.py

@@ -423,7 +423,7 @@ def set_secret(
         >>> parameters.set_secret(
                 name="my-secret",
                 value='{"password": "supers3cr3tllam@passw0rd"}',
-                client_request_token="61f2af5f-5f75-44b1-a29f-0cc37af55b11"
+                client_request_token="YOUR_TOKEN_HERE"
             )
 
     URLs: