Adding AppSync events

Author: leandrodamascena

aws_lambda_powertools/event_handler/events_appsync/base.py

@@ -3,30 +3,32 @@
 from abc import ABC, abstractmethod
 from typing import Callable
 
+DEFAULT_ROUTE = "/default/*"
+
 
 class BaseRouter(ABC):
     """Abstract base class for Router (resolvers)"""
 
     @abstractmethod
     def on_publish(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
         aggregate: bool = True,
     ) -> Callable:
         raise NotImplementedError
 
     @abstractmethod
     def async_on_publish(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
         aggregate: bool = True,
     ) -> Callable:
         raise NotImplementedError
 
     @abstractmethod
     def on_subscribe(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
     ) -> Callable:
         raise NotImplementedError
 

aws_lambda_powertools/event_handler/events_appsync/functions.py

@@ -35,9 +35,7 @@ def is_valid_path(path: str) -> bool:
         >>> is_valid_path('users')
         False
     """
-    if path == "/*":
-        return True
-    return bool(PATH_REGEX.fullmatch(path))
+    return True if path == "/*" else bool(PATH_REGEX.fullmatch(path))
 
 
 def find_best_route(routes: dict[str, Any], path: str):

aws_lambda_powertools/event_handler/events_appsync/router.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING
 
 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.event_handler.events_appsync.base import DEFAULT_ROUTE, BaseRouter
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -71,7 +71,7 @@ def __init__(self):
 
     def on_publish(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
         aggregate: bool = False,
     ) -> Callable:
         """
@@ -113,7 +113,7 @@ def on_publish(
 
     def async_on_publish(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
         aggregate: bool = False,
     ) -> Callable:
         """
@@ -154,7 +154,7 @@ def async_on_publish(
 
     def on_subscribe(
         self,
-        path: str = "/default/*",
+        path: str = DEFAULT_ROUTE,
     ) -> Callable:
         """
         Register a resolver function for subscribe operations.

tests/functional/event_handler/required_dependencies/appsync/test_appsync_events_resolvers.py

@@ -1041,7 +1041,7 @@ async def process_order(order_id, product_id, quantity):
             return {
                 "order_id": order_id,
                 "status": "processed",
-                "total_amount": quantity * 10.99,
+                "total_amount": quantity * 10,
             }
 
     order_service = MockOrderService()
@@ -1070,7 +1070,7 @@ async def process_order(payload):
     assert result["events"][0]["payload"]["order_processed"] is True
     assert result["events"][0]["payload"]["order_details"]["order_id"] == "order-123"
     assert result["events"][0]["payload"]["order_details"]["status"] == "processed"
-    assert result["events"][0]["payload"]["order_details"]["total_amount"] == 21.98  # 2 * 10.99
+    assert result["events"][0]["payload"]["order_details"]["total_amount"] == 20  # 2 * 10
 
 
 def test_complex_resolver_hierarchy(lambda_context, mock_event):