Adding routing script

Author: leandrodamascena

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Any, Callable
 
+from aws_lambda_powertools.event_handler.events_appsync.functions import find_best_route
+
 logger = logging.getLogger(__name__)
 
 
@@ -55,7 +57,7 @@ def find_resolver(self, path: str) -> dict | None:
             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}")
+        return self.resolvers.get(find_best_route(self.resolvers, path))
 
     def merge(self, other_registry: ResolverEventsRegistry):
         """Update current registry with incoming registry

aws_lambda_powertools/event_handler/events_appsync/functions.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+from functools import lru_cache
+from typing import Any
+
+
+def find_best_route(routes: dict[str, Any], path: str):
+    """
+    Find the most specific matching route for a given path.
+
+    Examples of matches:
+        Route: /default/v1/*         Path: /default/v1/users      -> MATCH
+        Route: /default/v1/*         Path: /default/v1/users/students  -> MATCH
+        Route: /default/v1/users/*   Path: /default/v1/users/123  -> MATCH (this wins over /default/v1/*)
+        Route: /*                    Path: /anything/here      -> MATCH (lowest priority)
+
+    Parameters
+    ----------
+    routes: dict[str, Any]
+        Dictionary containing routes and their handlers
+            Format: {
+                'resolvers': {
+                    '/path/*': {'func': callable, 'aggregate': bool},
+                    '/path/specific/*': {'func': callable, 'aggregate': bool}
+                }
+            }
+    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):
+        """
+        Convert a route pattern to a regex pattern with caching.
+        Examples:
+            /default/v1/*         -> ^/default/v1/[^/]+$
+            /default/v1/users/*   -> ^/default/v1/users/.*$
+
+        Parameters
+        ----------
+            route (str): Route pattern with wildcards
+
+        Returns
+        -------
+            Pattern: Compiled regex pattern
+        """
+        # Escape special regex chars but convert * to regex pattern
+        pattern = re.escape(route).replace("\\*", "[^/]+")
+
+        # If pattern ends with [^/]+, replace with .* for multi-segment match
+        if pattern.endswith("[^/]+"):
+            pattern = pattern[:-6] + ".*"
+
+        # Compile and return the regex pattern
+        return re.compile(f"^{pattern}$")
+
+    # Find all matching routes
+    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:
+    # - '/default/v1/users'     -> score: 14 (len=14, wildcards=0)
+    # - '/default/v1/users/*'   -> score: 14 (len=15, wildcards=1)
+    # - '/default/v1/*'        -> score: 8  (len=9, wildcards=1)
+    # - '/*'               -> score: 0  (len=2, wildcards=1)
+    return max(matches, key=lambda x: len(x) - x.count("*"), default=None)