Work work work..

Author: leandrodamascena

aws_lambda_powertools/event_handler/appsync_events.py

@@ -3,16 +3,22 @@
 import asyncio
 import logging
 import warnings
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.event_handler.events_appsync.router import Router
 from aws_lambda_powertools.utilities.data_classes.appsync_resolver_events_event import AppSyncResolverEventsEvent
 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
     from aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext
 
+
 logger = logging.getLogger(__name__)
+
+
 class AppSyncEventsResolver(Router):
     """
     AppSync Events API Resolver
@@ -25,35 +31,37 @@ def __init__(self):
 
     def __call__(
         self,
-        event: dict,
+        event: dict | AppSyncResolverEventsEvent,
         context: LambdaContext,
     ) -> Any:
         """Implicit lambda handler which internally calls `resolve`"""
         return self.resolve(event, context)
 
     def resolve(
         self,
-        event: AppSyncResolverEventsEvent,
+        event: dict | AppSyncResolverEventsEvent,
         context: LambdaContext,
     ) -> Any:
         """Resolves the response based on the provide event and decorator operation and namespaces"""
 
         self.lambda_context = context
         Router.lambda_context = context
 
-        Router.current_event = AppSyncResolverEventsEvent(event)
+        Router.current_event = (
+            event if isinstance(event, AppSyncResolverEventsEvent) else AppSyncResolverEventsEvent(event)
+        )
         self.current_event = Router.current_event
 
         if self.current_event.info.operation == "PUBLISH":
-            return self._call_publish_events(payload=self.current_event.events)
+            return self._publish_events(payload=self.current_event.events)
 
-        response = self._call_subscribe_events()
+        response = self._subscribe_events()
 
         self.clear_context()
 
         return response
 
-    def _call_subscribe_events(self) -> Any:
+    def _subscribe_events(self) -> Any:
         logger.debug(f"Processing subscribe events for path {self.current_event.info.channel_path}")
 
         resolver = self._subscribe_registry.find_resolver(self.current_event.info.channel_path)
@@ -66,7 +74,7 @@ def _call_subscribe_events(self) -> Any:
             return
         pass
 
-    def _call_publish_events(self, payload: list[dict[str, Any]]) -> Any:
+    def _publish_events(self, payload: list[dict[str, Any]]) -> list[dict[str, Any]] | dict[str, Any]:
         """Call single event resolver
 
         Parameters
@@ -90,34 +98,33 @@ def _call_publish_events(self, payload: list[dict[str, Any]]) -> Any:
 
         if resolver:
             logger.debug(f"Found sync resolver. {resolver}")
-            return self._call_publish_event_sync_resolver(
-                resolver=resolver["func"],
-                aggregate=resolver["aggregate"],
+            return self._process_publish_event_sync_resolver(
+                resolver=resolver,
             )
 
         if async_resolver:
             logger.debug(f"Found async resolver. {resolver}")
             return asyncio.run(
                 self._call_publish_event_async_resolver(
-                    resolver=async_resolver["func"],
-                    aggregate=async_resolver["aggregate"],
+                    resolver=async_resolver,
                 ),
             )
 
         # No resolver found
         # Warning and returning AS IS
         warnings.warn(
-            f"No resolvers were found for publish operations with path {self.current_event.info.channel_path}",
+            f"No resolvers were found for publish operations with path {self.current_event.info.channel_path}"
+            "We will return the entire payload as is",
             stacklevel=2,
-            category=PowertoolsUserWarning)
+            category=PowertoolsUserWarning,
+        )
 
         return {"events": payload}
 
-    def _call_publish_event_sync_resolver(
+    def _process_publish_event_sync_resolver(
         self,
-        resolver: Callable,
-        aggregate: bool = True,
-    ) -> list[Any]:
+        resolver: ResolverTypeDef,
+    ) -> list[dict[str, Any]] | dict[str, Any]:
         """
         Calls a synchronous batch resolver function for each event in the current batch.
 
@@ -140,34 +147,37 @@ def _call_publish_event_sync_resolver(
         """
 
         # Checks whether the entire batch should be processed at once
-        if aggregate:
-            # Process the entire batch
-            response = resolver(payload=self.current_event.events)
+        if resolver["aggregate"]:
+            try:
+                # Process the entire batch
+                response = resolver["func"](payload=self.current_event.events)
 
-            if not isinstance(response, list):
-                warnings.warn(
-                    "Response must be a list when using aggregate, AppSync will drop those events.",
-                    stacklevel=2,
-                    category=PowertoolsUserWarning)
-
-            return response
+                if not isinstance(response, list):
+                    warnings.warn(
+                        "Response must be a list when using aggregate, AppSync will drop those events.",
+                        stacklevel=2,
+                        category=PowertoolsUserWarning,
+                    )
 
+                return response
+            except Exception as error:
+                return {"error": self.format_error_response(error)}
 
         # By default, we gracefully append `None` for any records that failed processing
         results = []
         for idx, event in enumerate(self.current_event.events):
             try:
-                results.append(resolver(payload=event))
-            except Exception:
+                results.append(resolver["func"](payload=event))
+            except Exception as error:
                 logger.debug(f"Failed to process event number {idx}")
-                results.append(None)
+                error_return = {"id": event.get("id"), "error": self.format_error_response(error)}
+                results.append(error_return)
 
         return results
 
     async def _call_publish_event_async_resolver(
         self,
-        resolver: Callable,
-        aggregate: bool = True,
+        resolver: ResolverTypeDef,
     ) -> list[Any]:
         """
         Asynchronously call a batch resolver for each event in the current batch.
@@ -191,28 +201,55 @@ async def _call_publish_event_async_resolver(
         """
 
         # Checks whether the entire batch should be processed at once
-        if aggregate:
+        if resolver["aggregate"]:
             # Process the entire batch
-            response = await resolver(event=self.current_batch_event)
+            response = await resolver["func"](event=self.current_event.events)
             if not isinstance(response, list):
                 warnings.warn(
                     "Response must be a list when using aggregate, AppSync will drop those events.",
                     stacklevel=2,
-                    category=PowertoolsUserWarning)
+                    category=PowertoolsUserWarning,
+                )
 
             return response
 
-        response: list = []
+        response_async: list = []
 
         # Prime coroutines
-        tasks = [resolver(event=e, **e.arguments) for e in self.current_batch_event]
+        tasks = [resolver["func"](event=e) for e in self.current_event.events]
 
         # Aggregate results and exceptions, then filter them out
         # Use `None` upon exception for graceful error handling at GraphQL engine level
         #
         # NOTE: asyncio.gather(return_exceptions=True) catches and includes exceptions in the results
         #       this will become useful when we support exception handling in AppSync resolver
         results = await asyncio.gather(*tasks, return_exceptions=True)
-        response.extend(None if isinstance(ret, Exception) else ret for ret in results)
+        response_async.extend(None if isinstance(ret, Exception) else ret for ret in results)
 
-        return response
+        return response_async
+
+    def include_router(self, router: Router) -> None:
+        """Adds all resolvers defined in a router
+
+        Parameters
+        ----------
+        router : Router
+            A router containing a dict of field resolvers
+        """
+
+        # Merge app and router context
+        logger.debug("Merging router and app context")
+        self.context.update(**router.context)
+
+        # use pointer to allow context clearance after event is processed e.g., resolve(evt, ctx)
+        router.context = self.context
+
+        logger.debug("Merging router resolver registries")
+        self._publish_registry.merge(router._publish_registry)
+        self._async_publish_registry.merge(router._async_publish_registry)
+        self._subscribe_registry.merge(router._subscribe_registry)
+
+    def format_error_response(self, error=None) -> str:
+        if isinstance(error, Exception):
+            return f"{error.__class__.__name__} - {str(error)}"
+        return "An unknown error occurred"

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -2,18 +2,24 @@
 
 import logging
 import warnings
-from typing import Any, Callable, Literal
+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__)
 
-KIND_EVENT = Literal["on_publish", "async_on_publish", "on_subscribe"]
 
 class ResolverEventsRegistry:
+
     def __init__(self, kind_resolver: str):
-        self.resolvers: dict[str, dict[str, Any]] = {}
+        self.resolvers: dict[str, ResolverTypeDef] = {}
         self.kind_resolver = kind_resolver
 
     def register(
@@ -41,14 +47,14 @@ def register(
             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 '/'." 
+                "A path can have multiple namespaces, all separated by '/'."
                 "Wildcards are allowed only at the end of the path.",
                 stacklevel=2,
                 category=PowertoolsUserWarning,
             )
 
-
         def _register(func) -> Callable:
+
             print(f"Adding resolver `{func.__name__}` for path `{path}` and kind_resolver `{self.kind_resolver}`")
             self.resolvers[f"{path}"] = {
                 "func": func,
@@ -58,7 +64,7 @@ def _register(func) -> Callable:
 
         return _register
 
-    def find_resolver(self, path: str) -> dict | None:
+    def find_resolver(self, path: str) -> ResolverTypeDef | None:
         """Find resolver based on type_name and field_name
 
         Parameters

aws_lambda_powertools/event_handler/events_appsync/functions.py

@@ -6,6 +6,7 @@
 
 PATH_REGEX = re.compile(r"^\/([^\/\*]+)(\/[^\/\*]+)*(\/\*)?$")
 
+
 def is_valid_path(path: str) -> bool:
     """
     Checks if a given path is valid based on specific rules.
@@ -17,7 +18,7 @@ def is_valid_path(path: str) -> bool:
 
     Returns:
     --------
-    bool: 
+    bool:
         True if the path is valid, False otherwise
 
     Examples:
@@ -38,6 +39,7 @@ def is_valid_path(path: str) -> bool:
         return True
     return bool(PATH_REGEX.fullmatch(path))
 
+
 def find_best_route(routes: dict[str, Any], path: str):
     """
     Find the most specific matching route for a given path.
@@ -58,13 +60,14 @@ def find_best_route(routes: dict[str, Any], path: str):
                     '/path/specific/*': {'func': callable, 'aggregate': bool}
                 }
             }
-    path: str 
+    path: str
         Actual path to match (e.g., '/default/v1/users')
 
     Returns
     -------
         str: Most specific matching route or None if no match
     """
+
     @lru_cache(maxsize=1024)
     def pattern_to_regex(route):
         """
@@ -80,7 +83,7 @@ def pattern_to_regex(route):
 
         Returns
         -------
-        Pattern: 
+        Pattern:
             Compiled regex pattern
         """
         # Escape special regex chars but convert * to regex pattern
@@ -94,10 +97,7 @@ def pattern_to_regex(route):
         return re.compile(f"^{pattern}$")
 
     # Find all matching routes
-    matches = [
-        route for route in routes.keys()
-        if pattern_to_regex(route).match(path)
-    ]
+    matches = [route for route in routes.keys() if pattern_to_regex(route).match(path)]
 
     # Return the most specific route (longest length minus wildcards)
     # Examples of specificity:

aws_lambda_powertools/event_handler/events_appsync/router.py

@@ -13,8 +13,8 @@
 class Router(BaseRouter):
 
     context: dict
-    current_event: AppSyncResolverEventsEvent | None = None
-    lambda_context: LambdaContext | None = None
+    current_event: AppSyncResolverEventsEvent
+    lambda_context: LambdaContext
 
     def __init__(self):
         self.context = {}  # early init as customers might add context before event resolution