fix(api-gateway): non-greedy route pattern regex (aws-powertools#533)

aws_lambda_powertools/event_handler/api_gateway.py

@@ -20,6 +20,9 @@
 
 logger = logging.getLogger(__name__)
 
+_DYNAMIC_ROUTE_PATTERN = r"(<\w+>)"
+_NAMED_GROUP_BOUNDARY_PATTERN = r"(?P\1\\w+\\b)"
+
 
 class ProxyEventType(Enum):
     """An enumerations of the supported proxy event types."""
@@ -460,8 +463,35 @@ def __call__(self, event, context) -> Any:
 
     @staticmethod
     def _compile_regex(rule: str):
-        """Precompile regex pattern"""
-        rule_regex: str = re.sub(r"(<\w+>)", r"(?P\1.+)", rule)
+        """Precompile regex pattern
+
+        Logic
+        -----
+
+        1. Find any dynamic routes defined as <pattern>
+            e.g. @app.get("/accounts/<account_id>")
+        2. Create a new regex by substituting every dynamic route found as a named group (?P<group>),
+        and match whole words only (word boundary) instead of a greedy match
+
+            non-greedy example with word boundary
+
+                rule: '/accounts/<account_id>'
+                regex: r'/accounts/(?P<account_id>\\w+\\b)'
+
+                value: /accounts/123/some_other_path
+                account_id: 123
+
+            greedy example without word boundary
+
+                regex: r'/accounts/(?P<account_id>.+)'
+
+                value: /accounts/123/some_other_path
+                account_id: 123/some_other_path
+        3. Compiles a regex and include start (^) and end ($) in between for an exact match
+
+        NOTE: See #520 for context
+        """
+        rule_regex: str = re.sub(_DYNAMIC_ROUTE_PATTERN, _NAMED_GROUP_BOUNDARY_PATTERN, rule)
         return re.compile("^{}$".format(rule_regex))
 
     def _to_proxy_event(self, event: Dict) -> BaseProxyEvent:
@@ -485,7 +515,7 @@ def _resolve(self) -> ResponseBuilder:
             match: Optional[re.Match] = route.rule.match(path)
             if match:
                 logger.debug("Found a registered route. Calling function")
-                return self._call_route(route, match.groupdict())
+                return self._call_route(route, match.groupdict())  # pass fn args
 
         logger.debug(f"No match found for path {path} and method {method}")
         return self._not_found(method)

tests/functional/event_handler/test_api_gateway.py

@@ -1,6 +1,7 @@
 import base64
 import json
 import zlib
+from copy import deepcopy
 from decimal import Decimal
 from pathlib import Path
 from typing import Dict
@@ -664,3 +665,39 @@ def test_debug_print_event(capsys):
     # THEN print the event
     out, err = capsys.readouterr()
     assert json.loads(out) == event
+
+
+def test_similar_dynamic_routes():
+    # GIVEN
+    app = ApiGatewayResolver()
+    event = deepcopy(LOAD_GW_EVENT)
+
+    # WHEN
+    # r'^/accounts/(?P<account_id>\\w+\\b)$' # noqa: E800
+    @app.get("/accounts/<account_id>")
+    def get_account(account_id: str):
+        assert account_id == "single_account"
+
+    # r'^/accounts/(?P<account_id>\\w+\\b)/source_networks$' # noqa: E800
+    @app.get("/accounts/<account_id>/source_networks")
+    def get_account_networks(account_id: str):
+        assert account_id == "nested_account"
+
+    # r'^/accounts/(?P<account_id>\\w+\\b)/source_networks/(?P<network_id>\\w+\\b)$' # noqa: E800
+    @app.get("/accounts/<account_id>/source_networks/<network_id>")
+    def get_network_account(account_id: str, network_id: str):
+        assert account_id == "nested_account"
+        assert network_id == "network"
+
+    # THEN
+    event["resource"] = "/accounts/{account_id}"
+    event["path"] = "/accounts/single_account"
+    app.resolve(event, None)
+
+    event["resource"] = "/accounts/{account_id}/source_networks"
+    event["path"] = "/accounts/nested_account/source_networks"
+    app.resolve(event, None)
+
+    event["resource"] = "/accounts/{account_id}/source_networks/{network_id}"
+    event["path"] = "/accounts/nested_account/source_networks/network"
+    app.resolve(event, {})