feat(event-handler): add appsync batch resolvers

aws_lambda_powertools/event_handler/graphql_appsync/_registry.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)

aws_lambda_powertools/event_handler/graphql_appsync/base.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

aws_lambda_powertools/event_handler/graphql_appsync/exceptions.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
+    """

aws_lambda_powertools/event_handler/graphql_appsync/router.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()

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.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]:

aws_lambda_powertools/utilities/parameters/secrets.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: