Improve workflow

Author: leandrodamascena

aws_lambda_powertools/event_handler/api_gateway.py

@@ -6,7 +6,6 @@
 import logging
 import re
 import traceback
-import typing
 import warnings
 import zlib
 from abc import ABC, abstractmethod
@@ -468,10 +467,10 @@ def __init__(
 
         self.custom_response_validation_http_code = custom_response_validation_http_code
 
-        # _request_param_name caches the name of any Request-typed parameter in the handler (None = "not found").
-        # _request_param_checked avoids re-scanning the signature on every invocation.
-        self._request_param_name: str | None = None
-        self._request_param_name_checked: bool = False
+        # Caches the name of any Request-typed parameter in the handler.
+        # Avoids re-scanning the signature on every invocation.
+        self.request_param_name: str | None = None
+        self.request_param_name_checked: bool = False
 
     def __call__(
         self,
@@ -1638,16 +1637,23 @@ def my_middleware(app, next_middleware):
             return next_middleware(app)
         ```
         """
+        cached: Request | None = self.context.get("_request")
+        if cached is not None:
+            return cached
+
         route: Route | None = self.context.get("_route")
         if route is None:
             raise RuntimeError(
                 "app.request is only available after route resolution. Use it inside middleware or a route handler.",
             )
-        return Request(
+
+        request = Request(
             route_path=route.openapi_path,
             path_parameters=self.context.get("_route_args", {}),
             current_event=self.current_event,
         )
+        self.context["_request"] = request
+        return request
 
 
 class MiddlewareFrame:
@@ -1723,10 +1729,12 @@ def __call__(self, app: ApiGatewayResolver) -> dict | tuple | Response:
 
 def _find_request_param_name(func: Callable) -> str | None:
     """Return the name of the first parameter annotated as ``Request``, or ``None``."""
+    from typing import get_type_hints
+
     try:
         # get_type_hints resolves string annotations from ``from __future__ import annotations``
-        # using the function's own module globals — no pydantic dependency required.
-        hints = typing.get_type_hints(func)
+        # using the function's own module globals.
+        hints = get_type_hints(func)
     except Exception:
         hints = {}
 
@@ -1770,11 +1778,11 @@ def _registered_api_adapter(
     # Lookup is cached on the Route object to avoid repeated signature inspection.
     route: Route | None = app.context.get("_route")
     if route is not None:
-        if not route._request_param_name_checked:
-            route._request_param_name = _find_request_param_name(next_middleware)
-            route._request_param_name_checked = True
-        if route._request_param_name:
-            route_args = {**route_args, route._request_param_name: app.request}
+        if not route.request_param_name_checked:
+            route.request_param_name = _find_request_param_name(next_middleware)
+            route.request_param_name_checked = True
+        if route.request_param_name:
+            route_args = {**route_args, route.request_param_name: app.request}
 
     return app._to_response(next_middleware(**route_args))
 

tests/functional/event_handler/required_dependencies/test_request.py

@@ -515,3 +515,74 @@ def put_application(application_id: str):
     assert req.route == "/applications/{application_id}"
     assert req.path_parameters == {"application_id": "4da715ee-79d4-4e52-81cb-1ecc464708fb"}
     assert req.method == "PUT"
+
+
+# ---------------------------------------------------------------------------
+# Missing coverage: json_body, query_parameters=None, request caching
+# ---------------------------------------------------------------------------
+
+
+def test_request_json_body_in_middleware():
+    app = APIGatewayRestResolver()
+    bodies_seen: list = []
+
+    def mw(app: APIGatewayRestResolver, next_middleware):
+        bodies_seen.append(app.request.json_body)
+        return next_middleware(app)
+
+    app.use(middlewares=[mw])
+
+    @app.post("/items")
+    def handler():
+        return {}
+
+    event = _make_rest_event("/items", method="POST", body='{"name": "widget"}')
+    app(event, {})
+
+    assert bodies_seen == [{"name": "widget"}]
+
+
+def test_request_query_parameters_empty():
+    """When no query string parameters are present, query_parameters returns empty or None."""
+    app = APIGatewayRestResolver()
+    captured: list = []
+
+    def mw(app: APIGatewayRestResolver, next_middleware):
+        captured.append(app.request.query_parameters)
+        return next_middleware(app)
+
+    app.use(middlewares=[mw])
+
+    @app.get("/my/path")
+    def handler():
+        return {}
+
+    event = _make_rest_event("/my/path")
+    app(event, {})
+
+    # No query params present — should be falsy (empty dict or None depending on event source)
+    assert not captured[0]
+
+
+def test_request_is_cached_across_multiple_accesses():
+    """Accessing app.request multiple times in the same invocation returns the same object."""
+    app = APIGatewayRestResolver()
+    ids_seen: list[int] = []
+
+    def mw(app: APIGatewayRestResolver, next_middleware):
+        ids_seen.append(id(app.request))
+        ids_seen.append(id(app.request))
+        return next_middleware(app)
+
+    app.use(middlewares=[mw])
+
+    @app.get("/my/path")
+    def handler(request: Request):
+        ids_seen.append(id(request))
+        return {}
+
+    app(API_REST_EVENT, {})
+
+    # All accesses should return the same cached instance
+    assert len(ids_seen) == 3
+    assert ids_seen[0] == ids_seen[1] == ids_seen[2]