refactoring

Author: leandrodamascena

aws_lambda_powertools/event_handler/appsync_events.py

@@ -2,10 +2,12 @@
 
 import asyncio
 import logging
+import warnings
 from typing import TYPE_CHECKING, Any, Callable
 
 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 aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext
@@ -43,7 +45,7 @@ def resolve(
         self.current_event = Router.current_event
 
         if self.current_event.info.operation == "PUBLISH":
-            response = self._call_publish_events(payload=self.current_event.events)
+            return self._call_publish_events(payload=self.current_event.events)
 
         response = self._call_subscribe_events()
 
@@ -52,33 +54,165 @@ def resolve(
         return response
 
     def _call_subscribe_events(self) -> Any:
-        # PLACEHOLDER
+        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)
+        if not resolver:
+            warnings.warn(
+                f"No resolvers were found for publish operations with path {self.current_event.info.channel_path}",
+                stacklevel=2,
+                category=PowertoolsUserWarning,
+            )
+            return
         pass
 
     def _call_publish_events(self, payload: list[dict[str, Any]]) -> Any:
         """Call single event resolver
 
         Parameters
         ----------
-        event : dict
-            Event
-        data_model : type[AppSyncResolverEvent]
-            Data_model to decode AppSync event, by default it is of AppSyncResolverEvent type or subclass of it
+        payload : list[dict[str, Any]]
+            the messages sent by AppSync
         """
 
-        result = []
-        logger.debug("Processing direct resolver event")
+        logger.debug(f"Processing publish events for path {self.current_event.info.channel_path}")
 
-        #self.current_event = data_model(event)
         resolver = self._publish_registry.find_resolver(self.current_event.info.channel_path)
-        if not resolver:
-            print(f"No resolver found for '{self.current_event.info.channel_path}'")
-        print(resolver)
+        async_resolver = self._async_publish_registry.find_resolver(self.current_event.info.channel_path)
+
+        if resolver and async_resolver:
+            warnings.warn(
+                f"Both synchronous and asynchronous resolvers found for the same event and field."
+                f"The synchronous resolver takes precedence. Executing: {resolver['func'].__name__}",
+                stacklevel=2,
+                category=PowertoolsUserWarning,
+            )
+
+        if resolver:
+            logger.debug(f"Found sync resolver. {resolver}")
+            return self._call_publish_event_sync_resolver(
+                resolver=resolver["func"],
+                aggregate=resolver["aggregate"],
+            )
+
+        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"],
+                ),
+            )
+
+        # 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}",
+            stacklevel=2,
+            category=PowertoolsUserWarning)
+
+        return {"events": payload}
+
+    def _call_publish_event_sync_resolver(
+        self,
+        resolver: Callable,
+        aggregate: bool = True,
+    ) -> list[Any]:
+        """
+        Calls a synchronous batch resolver function for each event in the current batch.
+
+        Parameters
+        ----------
+        resolver: Callable
+            The callable function to resolve events.
+        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.
+
+        Returns
+        -------
+        list[Any]
+            A list of results corresponding to the resolved events.
+        """
+
+        # Checks whether the entire batch should be processed at once
+        if aggregate:
+            # Process the entire batch
+            response = resolver(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
+
+
+        # 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:
+                logger.debug(f"Failed to process event number {idx}")
+                results.append(None)
 
-        if not resolver["aggregate"]:
-            return resolver["func"](payload=self.current_event.events)
-        else:
-            for i in self.current_event.events:
-                result.append(resolver["func"](payload=i))
+        return results
 
-            return result
+    async def _call_publish_event_async_resolver(
+        self,
+        resolver: Callable,
+        aggregate: bool = True,
+    ) -> list[Any]:
+        """
+        Asynchronously call a batch resolver for each event in the current batch.
+
+        Parameters
+        ----------
+        resolver: Callable
+            The asynchronous resolver function.
+        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.
+
+        Returns
+        -------
+        list[Any]
+            A list of results corresponding to the resolved events.
+        """
+
+        # Checks whether the entire batch should be processed at once
+        if aggregate:
+            # Process the entire batch
+            response = await resolver(event=self.current_batch_event)
+            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
+
+        response: list = []
+
+        # Prime coroutines
+        tasks = [resolver(event=e, **e.arguments) for e in self.current_batch_event]
+
+        # 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)
+
+        return response

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -9,15 +9,17 @@
 
 logger = logging.getLogger(__name__)
 
+KIND_EVENT = Literal["on_publish", "async_on_publish", "on_subscribe"]
+
 class ResolverEventsRegistry:
-    def __init__(self):
+    def __init__(self, kind_resolver: str):
         self.resolvers: dict[str, dict[str, Any]] = {}
+        self.kind_resolver = kind_resolver
 
     def register(
         self,
         path: str = "/default/*",
         aggregate: bool = False,
-        operation: Literal["on_publish", "async_on_publish", "on_subscribe"] = "on_publish",
     ) -> Callable:
         """Registers the resolver for path that includes namespace + channel
 
@@ -27,8 +29,8 @@ def register(
             Path including namespace + channel
         aggregate: bool
             A flag indicating whether the batch items should be processed at once or individually.
-            If True , the batch resolver will process all items in the batch as a single event.
-            If False (default), the batch resolver will process each item in the batch individually.
+            If True, the resolver will process all items as a single event.
+            If False (default), the resolver will process each item individually.
 
         Return
         ----------
@@ -37,7 +39,7 @@ def register(
         """
         if not is_valid_path(path):
             warnings.warn(
-                f"The path `{path}` registered for `{operation}` is not valid and will be skipped."
+                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.",
@@ -47,7 +49,7 @@ def register(
 
 
         def _register(func) -> Callable:
-            logger.debug(f"Adding resolver `{func.__name__}` for path `{path}` and event `{operation}`")
+            print(f"Adding resolver `{func.__name__}` for path `{path}` and kind_resolver `{self.kind_resolver}`")
             self.resolvers[f"{path}"] = {
                 "func": func,
                 "aggregate": aggregate,
@@ -65,10 +67,10 @@ def find_resolver(self, path: str) -> dict | None:
             Type name
         Return
         ----------
-        Optional[Dict]
-            A dictionary with the resolver and if raise exception on error
+        dict | None
+            A dictionary with the resolver and if this is aggregated or not
         """
-        logger.debug(f"Looking for resolver for path={path}")
+        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):

aws_lambda_powertools/event_handler/events_appsync/router.py

@@ -4,10 +4,9 @@
 
 from aws_lambda_powertools.event_handler.events_appsync._registry import ResolverEventsRegistry
 from aws_lambda_powertools.event_handler.events_appsync.base import BaseRouter
-from aws_lambda_powertools.utilities.data_classes.appsync_resolver_events_event import AppSyncResolverEventsEvent
 
 if TYPE_CHECKING:
-    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncResolverEvent
+    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_events_event import AppSyncResolverEventsEvent
     from aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext
 
 
@@ -19,29 +18,29 @@ class Router(BaseRouter):
 
     def __init__(self):
         self.context = {}  # early init as customers might add context before event resolution
-        self._publish_registry = ResolverEventsRegistry()
-        self._async_publish_registry = ResolverEventsRegistry()
-        self._subscribe_registry = ResolverEventsRegistry()
+        self._publish_registry = ResolverEventsRegistry(kind_resolver="on_publish")
+        self._async_publish_registry = ResolverEventsRegistry(kind_resolver="async_on_publish")
+        self._subscribe_registry = ResolverEventsRegistry(kind_resolver="on_subscribe")
 
     def on_publish(
         self,
         path: str = "/default/*",
         aggregate: bool = False,
     ) -> Callable:
-        return self._publish_registry.register(path=path, aggregate=aggregate, operation="on_publish")
+        return self._publish_registry.register(path=path, aggregate=aggregate)
 
     def async_on_publish(
         self,
         path: str = "/default/*",
         aggregate: bool = False,
     ) -> Callable:
-        return self._async_publish_registry.register(path=path, aggregate=aggregate, operation="async_on_publish")
+        return self._async_publish_registry.register(path=path, aggregate=aggregate)
 
     def on_subscribe(
         self,
         path: str = "/default/*",
     ) -> Callable:
-        return self._subscribe_registry.register(path=path, operation="on_subscribe")
+        return self._subscribe_registry.register(path=path)
 
     def append_context(self, **additional_context):
         """Append key=value data as routing context"""