diff --git a/aws_lambda_powertools/event_handler/api_gateway.py b/aws_lambda_powertools/event_handler/api_gateway.py index 8b6c368af33..6948646a360 100644 --- a/aws_lambda_powertools/event_handler/api_gateway.py +++ b/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 + e.g. @app.get("/accounts/") + 2. Create a new regex by substituting every dynamic route found as a named group (?P), + and match whole words only (word boundary) instead of a greedy match + + non-greedy example with word boundary + + rule: '/accounts/' + regex: r'/accounts/(?P\\w+\\b)' + + value: /accounts/123/some_other_path + account_id: 123 + + greedy example without word boundary + + regex: r'/accounts/(?P.+)' + + 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) diff --git a/tests/functional/event_handler/test_api_gateway.py b/tests/functional/event_handler/test_api_gateway.py index b39dccc6084..41d2986d01e 100644 --- a/tests/functional/event_handler/test_api_gateway.py +++ b/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,40 @@ 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) + + # r'^/accounts/(?P\\w+\\b)$' # noqa: E800 + @app.get("/accounts/") + def get_account(account_id: str): + assert account_id == "single_account" + return {"message": f"{account_id}"} + + # r'^/accounts/(?P\\w+\\b)/source_networks$' # noqa: E800 + @app.get("/accounts//source_networks") + def get_account_networks(account_id: str): + assert account_id == "nested_account" + return {"message": f"{account_id}"} + + # r'^/accounts/(?P\\w+\\b)/source_networks/(?P\\w+\\b)$' # noqa: E800 + @app.get("/accounts//source_networks/") + def get_network_account(account_id: str, network_id: str): + assert account_id == "nested_account" + assert network_id == "network" + return {"message": f"{account_id}"} + + event["resource"] = "/accounts/{account_id}/source_networks" + event["path"] = "/accounts/nested_account/source_networks" + app.resolve(event, None) + + event["resource"] = "/accounts/{account_id}" + event["path"] = "/accounts/single_account" + app.resolve(event, None) + + event["resource"] = "/accounts/{account_id}/source_networks/{network_id}" + event["path"] = "/accounts/nested_account/source_networks/network" + app.resolve(event, {})