refactoring

Author: leandrodamascena

aws_lambda_powertools/event_handler/appsync_events.py

@@ -5,7 +5,6 @@
 from typing import TYPE_CHECKING, Any, Callable
 
 from aws_lambda_powertools.event_handler.events_appsync.router import Router
-from aws_lambda_powertools.event_handler.exception_handling import ExceptionHandlerManager
 from aws_lambda_powertools.utilities.data_classes.appsync_resolver_events_event import AppSyncResolverEventsEvent
 
 if TYPE_CHECKING:
@@ -20,52 +19,43 @@ class AppSyncEventsResolver(Router):
     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__(
         self,
         event: dict,
         context: LambdaContext,
-        data_model: type[AppSyncResolverEventsEvent] = AppSyncResolverEventsEvent,
     ) -> Any:
         """Implicit lambda handler which internally calls `resolve`"""
-        return self.resolve(event, context, data_model)
+        return self.resolve(event, context)
 
     def resolve(
         self,
         event: AppSyncResolverEventsEvent,
         context: LambdaContext,
-        data_model: type[AppSyncResolverEventsEvent] = AppSyncResolverEventsEvent,
     ) -> 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 = data_model(event)
-        self.current_event = data_model(event)
+        Router.current_event = AppSyncResolverEventsEvent(event)
+        self.current_event = Router.current_event
 
-        try:
-            if self.current_event.info.operation == "PUBLISH":
-                response = self._call_publish_events(payload=self.current_event.events, data_model=data_model)
-            else:
-                response = self._call_subscribe_events(event=event, data_model=data_model)
-        except Exception as exp:
-            response_builder = self.exception_handler_manager.lookup_exception_handler(type(exp))
-            if response_builder:
-                return response_builder(exp)
-            raise
+        if self.current_event.info.operation == "PUBLISH":
+            response = self._call_publish_events(payload=self.current_event.events)
+
+        response = self._call_subscribe_events()
 
         self.clear_context()
 
         return response
 
-    def _call_subscribe_events(self, payload: list[dict[str, Any]], data_model: type[AppSyncResolverEventsEvent]) -> Any:
+    def _call_subscribe_events(self) -> Any:
         # PLACEHOLDER
         pass
 
-    def _call_publish_events(self, payload: list[dict[str, Any]], data_model: type[AppSyncResolverEventsEvent]) -> Any:
+    def _call_publish_events(self, payload: list[dict[str, Any]]) -> Any:
         """Call single event resolver
 
         Parameters
@@ -92,46 +82,3 @@ def _call_publish_events(self, payload: list[dict[str, Any]], data_model: type[A
                 result.append(resolver["func"](payload=i))
 
             return result
-
-    def on_publish(
-        self,
-        path: str = "/default/*",
-        aggregate: bool = True,
-    ) -> Callable:
-        return self._publish_registry.register(path=path, aggregate=aggregate)
-
-    def async_on_publish(
-        self,
-        path: str = "/default/*",
-        aggregate: bool = True,
-    ) -> Callable:
-        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)
-
-    def async_on_subscribe(
-        self,
-        path: str = "/default/*",
-    ) -> Callable:
-        return self._async_subscribe_registry.register(path=path)
-
-    def exception_handler(self, exc_class: type[Exception] | list[type[Exception]]):
-        """
-        A decorator function that registers a handler for one or more exception types.
-
-        Parameters
-        ----------
-        exc_class (type[Exception] | list[type[Exception]])
-            A single exception type or a list of exception types.
-
-        Returns
-        -------
-        Callable:
-            A decorator function that registers the exception handler.
-        """
-
-        return self.exception_handler_manager.exception_handler(exc_class=exc_class)

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -1,21 +1,23 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Callable
+import warnings
+from typing import Any, Callable, Literal
 
-from aws_lambda_powertools.event_handler.events_appsync.functions import find_best_route
+from aws_lambda_powertools.event_handler.events_appsync.functions import find_best_route, is_valid_path
+from aws_lambda_powertools.warnings import PowertoolsUserWarning
 
 logger = logging.getLogger(__name__)
 
-
 class ResolverEventsRegistry:
     def __init__(self):
         self.resolvers: dict[str, dict[str, Any]] = {}
 
     def register(
         self,
         path: str = "/default/*",
-        aggregate: bool = True,
+        aggregate: bool = False,
+        operation: Literal["on_publish", "async_on_publish", "on_subscribe"] = "on_publish",
     ) -> Callable:
         """Registers the resolver for path that includes namespace + channel
 
@@ -25,17 +27,27 @@ def register(
             Path including namespace + channel
         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.
+            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.
 
         Return
         ----------
         Callable
             A Callable
         """
+        if not is_valid_path(path):
+            warnings.warn(
+                f"The path `{path}` registered for `{operation}` 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,
+            )
+
 
         def _register(func) -> Callable:
-            logger.debug(f"Adding resolver `{func.__name__}` for path `{path}`")
+            logger.debug(f"Adding resolver `{func.__name__}` for path `{path}` and event `{operation}`")
             self.resolvers[f"{path}"] = {
                 "func": func,
                 "aggregate": aggregate,

aws_lambda_powertools/event_handler/events_appsync/base.py

@@ -30,13 +30,6 @@ def on_subscribe(
     ) -> Callable:
         raise NotImplementedError
 
-    @abstractmethod
-    def async_on_subscribe(
-        self,
-        path: str = "/default/*",
-    ) -> Callable:
-        raise NotImplementedError
-    @abstractmethod
     def append_context(self, **additional_context) -> None:
         """
         Appends context information available under any route.

aws_lambda_powertools/event_handler/events_appsync/functions.py

@@ -4,6 +4,39 @@
 from functools import lru_cache
 from typing import Any
 
+PATH_REGEX = re.compile(r"^\/([^\/\*]+)(\/[^\/\*]+)*(\/\*)?$")
+
+def is_valid_path(path: str) -> bool:
+    """
+    Checks if a given path is valid based on specific rules.
+
+    Parameters
+    ----------
+    path: str
+        The path to validate
+
+    Returns:
+    --------
+    bool: 
+        True if the path is valid, False otherwise
+
+    Examples:
+        >>> is_valid_path('/*')
+        True
+        >>> is_valid_path('/users')
+        True
+        >>> is_valid_path('/users/profile')
+        True
+        >>> is_valid_path('/users/*/details')
+        False
+        >>> is_valid_path('/users/*')
+        True
+        >>> is_valid_path('users')
+        False
+    """
+    if path == "/*":
+        return True
+    return bool(PATH_REGEX.fullmatch(path))
 
 def find_best_route(routes: dict[str, Any], path: str):
     """
@@ -42,11 +75,13 @@ def pattern_to_regex(route):
 
         Parameters
         ----------
-            route (str): Route pattern with wildcards
+        route: str
+            Route pattern with wildcards
 
         Returns
         -------
-            Pattern: Compiled regex pattern
+        Pattern: 
+            Compiled regex pattern
         """
         # Escape special regex chars but convert * to regex pattern
         pattern = re.escape(route).replace("\\*", "[^/]+")

aws_lambda_powertools/event_handler/events_appsync/router.py

@@ -22,33 +22,26 @@ def __init__(self):
         self._publish_registry = ResolverEventsRegistry()
         self._async_publish_registry = ResolverEventsRegistry()
         self._subscribe_registry = ResolverEventsRegistry()
-        self._async_subscribe_registry = ResolverEventsRegistry()
 
     def on_publish(
         self,
         path: str = "/default/*",
-        aggregate: bool = True,
+        aggregate: bool = False,
     ) -> Callable:
-        return self._publish_registry.register(path=path, aggregate=aggregate)
+        return self._publish_registry.register(path=path, aggregate=aggregate, operation="on_publish")
 
     def async_on_publish(
         self,
         path: str = "/default/*",
-        aggregate: bool = True,
+        aggregate: bool = False,
     ) -> Callable:
-        return self._async_publish_registry.register(path=path, aggregate=aggregate)
+        return self._async_publish_registry.register(path=path, aggregate=aggregate, operation="async_on_publish")
 
     def on_subscribe(
         self,
         path: str = "/default/*",
     ) -> Callable:
-        return self._subscribe_registry.register(path=path)
-
-    def async_on_subscribe(
-        self,
-        path: str = "/default/*",
-    ) -> Callable:
-        return self._async_subscribe_registry.register(path=path)
+        return self._subscribe_registry.register(path=path, operation="on_subscribe")
 
     def append_context(self, **additional_context):
         """Append key=value data as routing context"""