fix(event_handlers): Bedrock kwargs and media type, AppSync registry and context, Router duplicates, Function URL base path, Depends with classes

aws_lambda_powertools/event_handler/api_gateway.py

@@ -3333,6 +3333,12 @@ def register_route(func: AnyCallableT) -> AnyCallableT:
                 status_code,
             )
 
+            if route_key in self._routes:
+                warnings.warn(
+                    f"A route like this was already registered. method: '{', '.join(methods)}' rule: '{rule}'",
+                    stacklevel=2,
+                )
+
             # Collate Middleware for routes
             if middlewares is not None:
                 for handler in middlewares:

aws_lambda_powertools/event_handler/appsync.py

@@ -147,6 +147,7 @@ def lambda_handler(event, context):
         self.lambda_context = context
         Router.lambda_context = context
 
+        response = None
         try:
             if isinstance(event, list):
                 Router.current_batch_event = [data_model(e) for e in event]
@@ -157,16 +158,17 @@ def lambda_handler(event, context):
         except Exception as exp:
             response_builder = self.exception_handler_manager.lookup_exception_handler(type(exp))
             if response_builder:
-                return response_builder(exp)
-            raise
-
-        # We don't clear the context for coroutines because we don't have control over the event loop.
-        # If we clean the context immediately, it might not be available when the coroutine is actually executed.
-        # For single async operations, the context should be cleaned up manually after the coroutine completes.
-        # See: https://github.com/aws-powertools/powertools-lambda-python/issues/5290
-        # REVIEW: Review this support in Powertools V4
-        if not asyncio.iscoroutine(response):
-            self.clear_context()
+                response = response_builder(exp)
+            else:
+                raise
+        finally:
+            # We don't clear the context for coroutines because we don't have control over the event loop.
+            # If we clean the context immediately, it might not be available when the coroutine is actually executed.
+            # For single async operations, the context should be cleaned up manually after the coroutine completes.
+            # See: https://github.com/aws-powertools/powertools-lambda-python/issues/5290
+            # REVIEW: Review this support in Powertools V4
+            if not asyncio.iscoroutine(response):
+                self.clear_context()
 
         return response
 

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -49,7 +49,7 @@ def build(self, event: BedrockAgentEvent, *args) -> dict[str, Any]:
                 "httpMethod": event.http_method,
                 "httpStatusCode": self.response.status_code,
                 "responseBody": {
-                    self.response.content_type: {
+                    self.response.content_type or "application/json": {
                         "body": body,
                     },
                 },
@@ -346,8 +346,13 @@ def _convert_matches_into_route_keys(self, match: Match) -> dict[str, str]:
         # In Bedrock Agents, all the parameters come inside the "parameters" key, not on the apiPath
         # So we have to search for route parameters in the parameters key
         parameters: dict[str, str] = {}
-        if match.groupdict() and self.current_event.parameters:
-            parameters = {parameter["name"]: parameter["value"] for parameter in self.current_event.parameters}
+        route_keys = match.groupdict()
+        if route_keys and self.current_event.parameters:
+            parameters = {
+                parameter["name"]: parameter["value"]
+                for parameter in self.current_event.parameters
+                if parameter["name"] in route_keys
+            }
         return parameters
 
     @override

aws_lambda_powertools/event_handler/depends.py

@@ -114,6 +114,11 @@ def _get_depends_from_annotation(annotation: Any) -> Depends | None:
     return None
 
 
+def _annotation_target(func: Callable[..., Any]) -> Callable[..., Any]:
+    """A class carries its dependency annotations on __init__, not on the class itself."""
+    return func.__init__ if isinstance(func, type) else func  # type: ignore[misc]
+
+
 def _has_depends(func: Callable[..., Any]) -> bool:
     """Check if a callable has any Depends() parameters, without importing pydantic."""
     try:
@@ -134,7 +139,7 @@ def build_dependency_tree(func: Callable[..., Any]) -> DependencyTree:
     annotations and recursively builds the tree — all without importing pydantic.
     """
     try:
-        hints = get_type_hints(func, include_extras=True)
+        hints = get_type_hints(_annotation_target(func), include_extras=True)
     except Exception:
         return DependencyTree()
 
@@ -214,7 +219,7 @@ def solve_dependencies(
         # Inject Request if the dependency declares it
         if request is not None:
             try:
-                hints = get_type_hints(use_fn)
+                hints = get_type_hints(_annotation_target(use_fn))
             except Exception:  # pragma: no cover - defensive for broken annotations
                 hints = {}
             for param_name, annotation in hints.items():

aws_lambda_powertools/event_handler/events_appsync/_registry.py

@@ -43,7 +43,7 @@ def register(
             A Callable
         """
 
-        def _register(func) -> Callable | None:
+        def _register(func) -> Callable:
             if not is_valid_path(path):
                 warnings.warn(
                     f"The path `{path}` registered for `{self.kind_resolver}` is not valid and will be skipped."
@@ -53,7 +53,7 @@ def _register(func) -> Callable | None:
                     stacklevel=2,
                     category=PowertoolsUserWarning,
                 )
-                return None
+                return func
 
             logger.debug(
                 f"Adding resolver `{func.__name__}` for path `{path}` and kind_resolver `{self.kind_resolver}`",

aws_lambda_powertools/event_handler/lambda_function_url.py

@@ -52,6 +52,6 @@ def lambda_handler(event, context):
 
     def _get_base_path(self) -> str:
         stage = self.current_event.request_context.stage
-        if stage and stage != "$default" and self.current_event.request_context.http.method.startswith(f"/{stage}"):
+        if stage and stage != "$default" and self.current_event.request_context.http.path.startswith(f"/{stage}"):
             return f"/{stage}"
         return ""

tests/functional/event_handler/_pydantic/test_bedrock_agent.py

@@ -418,3 +418,39 @@ def claims() -> dict[str, Any]:
     body = result["response"]["responseBody"]["application/json"]["body"]
     assert "잔액은 1,000원입니다 💰" in body
     assert json.loads(body) == {"output": "잔액은 1,000원입니다 💰"}
+
+
+def test_bedrock_agent_ignores_non_path_params_when_validation_is_off():
+    # GIVEN a Bedrock Agent event with a parameter that is not a path parameter
+    event = load_event("bedrockAgentEventWithPathParams.json")
+    event["parameters"] = [*event["parameters"], {"type": "string", "name": "verbose", "value": "true"}]
+
+    app = BedrockAgentResolver(enable_validation=False)
+
+    @app.get("/claims/<claim_id>", description="Gets claims by ID")
+    def claims(claim_id: str):
+        return {"claim_id": claim_id}
+
+    # WHEN calling the event handler
+    result = app(event, {})
+
+    # THEN only path parameters are injected as kwargs and the handler resolves
+    assert result["response"]["httpStatusCode"] == 200
+    body = json.loads(result["response"]["responseBody"]["application/json"]["body"])
+    assert body == {"claim_id": "123"}
+
+
+def test_bedrock_agent_response_without_content_type_defaults_to_json():
+    # GIVEN a handler returning a Response without a content type
+    app = BedrockAgentResolver(enable_validation=False)
+
+    @app.get("/claims", description="Gets claims")
+    def claims():
+        return Response(status_code=200, body="ok")
+
+    # WHEN calling the event handler
+    result = app(load_event("bedrockAgentEvent.json"), {})
+
+    # THEN the response body media type defaults to application/json instead of "null"
+    assert "null" not in result["response"]["responseBody"]
+    assert result["response"]["responseBody"]["application/json"]["body"] == "ok"

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

@@ -1643,3 +1643,19 @@ def handle_events():
     # WHEN we resolve the event with unauthorized route
     with pytest.raises(UnauthorizedException):
         app.resolve(mock_event, lambda_context)
+
+
+def test_invalid_path_registration_still_returns_function():
+    # GIVEN an AppSyncEventsResolver
+    app = AppSyncEventsResolver()
+
+    # WHEN registering a resolver for an invalid path
+    with pytest.warns(PowertoolsUserWarning, match="is not valid and will be skipped"):
+
+        @app.on_publish(path="/default/*/sub")
+        def handler(payload):
+            return payload
+
+    # THEN the decorated function is returned instead of None
+    assert handler is not None
+    assert handler("payload") == "payload"

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

@@ -387,3 +387,26 @@ def create_something(id: str):  # noqa AA03 VNE003
 
     # THEN the return must be the Exception Handler error message
     assert result["message"] == "error"
+
+
+def test_exception_handler_clears_context():
+    # GIVEN an AppSyncResolver with an exception handler and a resolver that appends context
+    mock_event = load_event("appSyncDirectResolver.json")
+
+    app = AppSyncResolver()
+
+    @app.exception_handler(ValueError)
+    def handle_value_error(ex: ValueError):
+        return {"error": str(ex)}
+
+    @app.resolver(field_name="createSomething")
+    def create_something(id: str):  # noqa AA03 VNE003
+        app.append_context(invocation="data")
+        raise ValueError("something went wrong")
+
+    # WHEN resolving an event that triggers the exception handler
+    result = app.resolve(mock_event, {})
+
+    # THEN the handler result is returned and the context does not leak into the next invocation
+    assert result == {"error": "something went wrong"}
+    assert app.context == {}

tests/functional/event_handler/required_dependencies/test_depends.py

@@ -647,6 +647,43 @@ def handler(user: Annotated[str, Depends(get_current_user)]):
     assert json.loads(result["body"]) == {"statusCode": 401, "message": "Missing authentication"}
 
 
+def test_depends_class_dependency_receives_request():
+    """A class dependency gets Request injected into its __init__."""
+    app = APIGatewayHttpResolver()
+
+    class Service:
+        def __init__(self, request: Request):
+            self.path = request.resolved_event.path
+
+    @app.post("/my/path")
+    def handler(svc: Annotated[Service, Depends(Service)]):
+        return {"path": svc.path}
+
+    result = app(API_GW_V2_EVENT, {})
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"path": "/my/path"}
+
+
+def test_depends_class_dependency_with_nested_depends():
+    """A class dependency resolves its own Depends() parameters declared on __init__."""
+    app = APIGatewayHttpResolver()
+
+    def get_tenant() -> str:
+        return "tenant-1"
+
+    class Service:
+        def __init__(self, tenant: Annotated[str, Depends(get_tenant)]):
+            self.tenant = tenant
+
+    @app.post("/my/path")
+    def handler(svc: Annotated[Service, Depends(Service)]):
+        return {"tenant": svc.tenant}
+
+    result = app(API_GW_V2_EVENT, {})
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == {"tenant": "tenant-1"}
+
+
 def test_depends_with_none_default_is_resolved():
     """A Depends() parameter that also declares ``= None`` is still resolved (Python 3.10 wraps it in Optional)."""
     app = APIGatewayHttpResolver()

tests/functional/event_handler/required_dependencies/test_lambda_function_url.py

@@ -82,3 +82,18 @@ def foo():
     # THEN process event correctly
     # AND return 404 because the event doesn't match any known route
     assert result["statusCode"] == 404
+
+
+def test_lambda_function_url_get_base_path_checks_http_path():
+    # GIVEN a Lambda Function Url event with a custom stage prefixing the path
+    event = load_event("lambdaFunctionUrlEvent.json")
+    event["requestContext"]["stage"] = "live"
+    event["rawPath"] = "/live/my/path"
+    event["requestContext"]["http"]["path"] = "/live/my/path"
+
+    app = LambdaFunctionUrlResolver()
+    app.current_event = LambdaFunctionUrlEvent(event)
+
+    # WHEN deriving the base path
+    # THEN it comes from the http path, not the http method
+    assert app._get_base_path() == "/live"

tests/functional/event_handler/required_dependencies/test_router.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import pytest
+
 from aws_lambda_powertools.event_handler import (
     ALBResolver,
     APIGatewayHttpResolver,
@@ -76,3 +78,20 @@ def foo():
     app.include_router(router)
     result = app(load_event("lambdaFunctionUrlEvent.json"), {})
     assert result["body"] == "routed"
+
+
+def test_router_warns_on_duplicate_route():
+    # GIVEN a Router with a registered route
+    router = APIGatewayRouter()
+
+    @router.get("/my/path")
+    def first():
+        return {"fn": "first"}
+
+    # WHEN registering the same route again
+    # THEN a warning is emitted like the resolver does
+    with pytest.warns(UserWarning, match="A route like this was already registered"):
+
+        @router.get("/my/path")
+        def second():
+            return {"fn": "second"}