Adding registry logic

Author: leandrodamascena

aws_lambda_powertools/event_handler/appsync_events.py

@@ -1,10 +1,56 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING, Any, Callable
+
+from aws_lambda_powertools.event_handler.events_appsync._registry import ResolverEventsRegistry
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_events_event import AppSyncResolverEventsEvent
+    from aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext
+
 
 class AppSyncEventsResolver:
     """
     AppSync Events API Resolver
     """
 
     def __init__(self):
-        pass
+        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._async_subscribe_registry = ResolverEventsRegistry()
+
+    def resolve(
+        self,
+        event: AppSyncResolverEventsEvent,
+        context: LambdaContext,
+    ) -> Any:
+        """Resolves the response based on the provide event and decorator operation and namespaces"""
+        print(self._publish_registry.__dict__)
+
+    def publish(
+        self,
+        path: str = "/default/*",
+        aggregate: bool = True,
+    ) -> Callable:
+        return self._publish_registry.register(path=path, aggregate=aggregate)
+
+    def async_publish(
+        self,
+        path: str = "/default/*",
+        aggregate: bool = True,
+    ) -> Callable:
+        return self._async_publish_registry.register(path=path, aggregate=aggregate)
+
+    def subscribe(
+        self,
+        path: str = "/default/*",
+    ) -> Callable:
+        return self._subscribe_registry.register(path=path)
+
+    def async_subscribe(
+        self,
+        path: str = "/default/*",
+    ) -> Callable:
+        return self._async_subscribe_registry.register(path=path)

aws_lambda_powertools/event_handler/events_appsync/__init__.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable
+
+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,
+    ) -> Callable:
+        """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 (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.
+
+        Return
+        ----------
+        Callable
+            A Callable
+        """
+
+        def _register(func) -> Callable:
+            logger.debug(f"Adding resolver `{func.__name__}` for path `{path}`")
+            self.resolvers[f"{path}"] = {
+                "func": func,
+                "aggregate": aggregate,
+            }
+            return func
+
+        return _register
+
+    def find_resolver(self, path: str) -> dict | None:
+        """Find resolver based on type_name and field_name
+
+        Parameters
+        ----------
+        path : str
+            Type name
+        Return
+        ----------
+        Optional[Dict]
+            A dictionary with the resolver and if raise exception on error
+        """
+        logger.debug(f"Looking for resolver for path={path}")
+        return self.resolvers.get(f"{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)