feat(event_handler): add AppSync events resolver (#6558)

* Adding AppSync events

* Adding AppSync events

Author: leandrodamascena

aws_lambda_powertools/event_handler/__init__.py

@@ -12,13 +12,15 @@
 )
 from aws_lambda_powertools.event_handler.appsync import AppSyncResolver
 from aws_lambda_powertools.event_handler.bedrock_agent import BedrockAgentResolver
+from aws_lambda_powertools.event_handler.events_appsync.appsync_events import AppSyncEventsResolver
 from aws_lambda_powertools.event_handler.lambda_function_url import (
     LambdaFunctionUrlResolver,
 )
 from aws_lambda_powertools.event_handler.vpc_lattice import VPCLatticeResolver, VPCLatticeV2Resolver
 
 __all__ = [
     "AppSyncResolver",
+    "AppSyncEventsResolver",
     "APIGatewayRestResolver",
     "APIGatewayHttpResolver",
     "ALBResolver",

aws_lambda_powertools/event_handler/api_gateway.py

@@ -17,6 +17,7 @@
 from typing_extensions import override
 
 from aws_lambda_powertools.event_handler import content_types
+from aws_lambda_powertools.event_handler.exception_handling import ExceptionHandlerManager
 from aws_lambda_powertools.event_handler.exceptions import NotFoundError, ServiceError
 from aws_lambda_powertools.event_handler.openapi.config import OpenAPIConfig
 from aws_lambda_powertools.event_handler.openapi.constants import (
@@ -1576,6 +1577,7 @@ def __init__(
         self.processed_stack_frames = []
         self._response_builder_class = ResponseBuilder[BaseProxyEvent]
         self.openapi_config = OpenAPIConfig()  # starting an empty dataclass
+        self.exception_handler_manager = ExceptionHandlerManager()
         self._has_response_validation_error = response_validation_error_http_code is not None
         self._response_validation_error_http_code = self._validate_response_validation_error_http_code(
             response_validation_error_http_code,
@@ -2498,7 +2500,7 @@ def not_found_handler():
                 return Response(status_code=204, content_type=None, headers=_headers, body="")
 
             # Customer registered 404 route? Call it.
-            custom_not_found_handler = self._lookup_exception_handler(NotFoundError)
+            custom_not_found_handler = self.exception_handler_manager.lookup_exception_handler(NotFoundError)
             if custom_not_found_handler:
                 return custom_not_found_handler(NotFoundError())
 
@@ -2571,26 +2573,10 @@ def not_found(self, func: Callable | None = None):
         return self.exception_handler(NotFoundError)(func)
 
     def exception_handler(self, exc_class: type[Exception] | list[type[Exception]]):
-        def register_exception_handler(func: Callable):
-            if isinstance(exc_class, list):  # pragma: no cover
-                for exp in exc_class:
-                    self._exception_handlers[exp] = func
-            else:
-                self._exception_handlers[exc_class] = func
-            return func
-
-        return register_exception_handler
-
-    def _lookup_exception_handler(self, exp_type: type) -> Callable | None:
-        # Use "Method Resolution Order" to allow for matching against a base class
-        # of an exception
-        for cls in exp_type.__mro__:
-            if cls in self._exception_handlers:
-                return self._exception_handlers[cls]
-        return None
+        return self.exception_handler_manager.exception_handler(exc_class=exc_class)
 
     def _call_exception_handler(self, exp: Exception, route: Route) -> ResponseBuilder | None:
-        handler = self._lookup_exception_handler(type(exp))
+        handler = self.exception_handler_manager.lookup_exception_handler(type(exp))
         if handler:
             try:
                 return self._response_builder_class(response=handler(exp), serializer=self._serializer, route=route)
@@ -2686,7 +2672,7 @@ def include_router(self, router: Router, prefix: str | None = None) -> None:
         self._router_middlewares = self._router_middlewares + router._router_middlewares
 
         logger.debug("Appending Router exception_handler into App exception_handler.")
-        self._exception_handlers.update(router._exception_handlers)
+        self.exception_handler_manager.update_exception_handlers(router._exception_handlers)
 
         # use pointer to allow context clearance after event is processed e.g., resolve(evt, ctx)
         router.context = self.context

aws_lambda_powertools/event_handler/appsync.py

@@ -5,6 +5,7 @@
 import warnings
 from typing import TYPE_CHECKING, Any
 
+from aws_lambda_powertools.event_handler.exception_handling import ExceptionHandlerManager
 from aws_lambda_powertools.event_handler.graphql_appsync.exceptions import InvalidBatchResponse, ResolverNotFoundError
 from aws_lambda_powertools.event_handler.graphql_appsync.router import Router
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
@@ -55,6 +56,7 @@ def __init__(self):
         """
         super().__init__()
         self.context = {}  # early init as customers might add context before event resolution
+        self.exception_handler_manager = ExceptionHandlerManager()
         self._exception_handlers: dict[type, Callable] = {}
 
     def __call__(
@@ -153,7 +155,7 @@ def lambda_handler(event, context):
                 Router.current_event = data_model(event)
                 response = self._call_single_resolver(event=event, data_model=data_model)
         except Exception as exp:
-            response_builder = self._lookup_exception_handler(type(exp))
+            response_builder = self.exception_handler_manager.lookup_exception_handler(type(exp))
             if response_builder:
                 return response_builder(exp)
             raise
@@ -495,31 +497,4 @@ def exception_handler(self, exc_class: type[Exception] | list[type[Exception]]):
             A decorator function that registers the exception handler.
         """
 
-        def register_exception_handler(func: Callable):
-            if isinstance(exc_class, list):  # pragma: no cover
-                for exp in exc_class:
-                    self._exception_handlers[exp] = func
-            else:
-                self._exception_handlers[exc_class] = func
-            return func
-
-        return register_exception_handler
-
-    def _lookup_exception_handler(self, exp_type: type) -> Callable | None:
-        """
-        Looks up the registered exception handler for the given exception type or its base classes.
-
-        Parameters
-        ----------
-        exp_type (type):
-            The exception type to look up the handler for.
-
-        Returns
-        -------
-        Callable | None:
-            The registered exception handler function if found, otherwise None.
-        """
-        for cls in exp_type.__mro__:
-            if cls in self._exception_handlers:
-                return self._exception_handlers[cls]
-        return None
+        return self.exception_handler_manager.exception_handler(exc_class=exc_class)

aws_lambda_powertools/event_handler/events_appsync/__init__.py

@@ -0,0 +1,5 @@
+from aws_lambda_powertools.event_handler.events_appsync.appsync_events import AppSyncEventsResolver
+
+__all__ = [
+    "AppSyncEventsResolver",
+]

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import logging
+import warnings
+from typing import TYPE_CHECKING
+
+from aws_lambda_powertools.event_handler.events_appsync.functions import find_best_route, is_valid_path
+from aws_lambda_powertools.warnings import PowertoolsUserWarning
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from aws_lambda_powertools.event_handler.events_appsync.types import ResolverTypeDef
+
+
+logger = logging.getLogger(__name__)
+
+
+class ResolverEventsRegistry:
+    def __init__(self, kind_resolver: str):
+        self.resolvers: dict[str, ResolverTypeDef] = {}
+        self.kind_resolver = kind_resolver
+
+    def register(
+        self,
+        path: str = "/default/*",
+        aggregate: bool = False,
+    ) -> Callable | None:
+        """Registers the resolver for path that includes namespace + channel
+
+        Parameters
+        ----------
+        path : str
+            Path including namespace + channel
+        aggregate: bool
+            A flag indicating whether the batch items should be processed at once or individually.
+            If True, the resolver will process all items as a single event.
+            If False (default), the resolver will process each item individually.
+
+        Return
+        ----------
+        Callable
+            A Callable
+        """
+
+        def _register(func) -> Callable | None:
+            if not is_valid_path(path):
+                warnings.warn(
+                    f"The path `{path}` registered for `{self.kind_resolver}` is not valid and will be skipped."
+                    f"A path should always have a namespace starting with '/'"
+                    "A path can have multiple namespaces, all separated by '/'."
+                    "Wildcards are allowed only at the end of the path.",
+                    stacklevel=2,
+                    category=PowertoolsUserWarning,
+                )
+                return None
+
+            logger.debug(
+                f"Adding resolver `{func.__name__}` for path `{path}` and kind_resolver `{self.kind_resolver}`",
+            )
+            self.resolvers[f"{path}"] = {
+                "func": func,
+                "aggregate": aggregate,
+            }
+            return func
+
+        return _register
+
+    def find_resolver(self, path: str) -> ResolverTypeDef | None:
+        """Find resolver based on type_name and field_name
+
+        Parameters
+        ----------
+        path : str
+            Type name
+        Return
+        ----------
+        dict | None
+            A dictionary with the resolver and if this is aggregated or not
+        """
+        logger.debug(f"Looking for resolver for path `{path}` and kind_resolver `{self.kind_resolver}`")
+        return self.resolvers.get(find_best_route(self.resolvers, path))
+
+    def merge(self, other_registry: ResolverEventsRegistry):
+        """Update current registry with incoming registry
+
+        Parameters
+        ----------
+        other_registry : ResolverRegistry
+            Registry to merge from
+        """
+        self.resolvers.update(**other_registry.resolvers)