feat(event_handler): add Request object for middleware access to resolved route and args

aws_lambda_powertools/event_handler/__init__.py

@@ -21,6 +21,7 @@
 from aws_lambda_powertools.event_handler.lambda_function_url import (
     LambdaFunctionUrlResolver,
 )
+from aws_lambda_powertools.event_handler.request import Request
 from aws_lambda_powertools.event_handler.vpc_lattice import VPCLatticeResolver, VPCLatticeV2Resolver
 
 __all__ = [
@@ -37,6 +38,7 @@
     "CORSConfig",
     "HttpResolverLocal",
     "LambdaFunctionUrlResolver",
+    "Request",
     "Response",
     "VPCLatticeResolver",
     "VPCLatticeV2Resolver",

aws_lambda_powertools/event_handler/api_gateway.py

@@ -6,6 +6,7 @@
 import logging
 import re
 import traceback
+import typing
 import warnings
 import zlib
 from abc import ABC, abstractmethod
@@ -20,6 +21,7 @@
 from aws_lambda_powertools.event_handler import content_types
 from aws_lambda_powertools.event_handler.exception_handling import ExceptionHandlerManager
 from aws_lambda_powertools.event_handler.exceptions import NotFoundError, ServiceError
+from aws_lambda_powertools.event_handler.request import Request
 from aws_lambda_powertools.event_handler.openapi.config import OpenAPIConfig
 from aws_lambda_powertools.event_handler.openapi.constants import (
     DEFAULT_API_VERSION,
@@ -466,6 +468,11 @@ 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
+
     def __call__(
         self,
         router_middlewares: list[Callable],
@@ -1608,6 +1615,41 @@ def clear_context(self):
         """Resets routing context"""
         self.context.clear()
 
+    @property
+    def request(self) -> Request:
+        """Current resolved :class:`Request` object.
+
+        Available inside middleware and in route handlers that declare a parameter
+        typed as :class:`Request <aws_lambda_powertools.event_handler.request.Request>`.
+
+        Raises
+        ------
+        RuntimeError
+            When accessed before route resolution (i.e. outside of middleware / handler scope).
+
+        Examples
+        --------
+        **Middleware**
+
+        ```python
+        def my_middleware(app, next_middleware):
+            req = app.request
+            print(req.route, req.method, req.path_parameters)
+            return next_middleware(app)
+        ```
+        """
+        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(
+            route_path=route.openapi_path,
+            path_parameters=self.context.get("_route_args", {}),
+            current_event=self.current_event,
+        )
+
 
 class MiddlewareFrame:
     """
@@ -1680,6 +1722,22 @@ def __call__(self, app: ApiGatewayResolver) -> dict | tuple | Response:
         return self.current_middleware(app, self.next_middleware)
 
 
+def _find_request_param_name(func: Callable) -> str | None:
+    """Return the name of the first parameter annotated as ``Request``, or ``None``."""
+    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)
+    except Exception:
+        hints = {}
+
+    for param_name, annotation in hints.items():
+        if annotation is Request:
+            return param_name
+
+    return None
+
+
 def _registered_api_adapter(
     app: ApiGatewayResolver,
     next_middleware: Callable[..., Any],
@@ -1708,6 +1766,17 @@ def _registered_api_adapter(
     """
     route_args: dict = app.context.get("_route_args", {})
     logger.debug(f"Calling API Route Handler: {route_args}")
+
+    # Inject a Request object when the handler declares a parameter typed as Request.
+    # 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}
+
     return app._to_response(next_middleware(**route_args))
 
 

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -4,6 +4,7 @@
 import re
 from typing import TYPE_CHECKING, Any, ForwardRef, cast
 
+from aws_lambda_powertools.event_handler.request import Request
 from aws_lambda_powertools.event_handler.openapi.compat import (
     ModelField,
     create_body_model,
@@ -187,6 +188,11 @@ def get_dependant(
 
     # Add each parameter to the dependant model
     for param_name, param in signature_params.items():
+        # Request-typed parameters are injected by the resolver at call time;
+        # they carry no OpenAPI meaning and must be excluded from schema generation.
+        if param.annotation is Request:
+            continue
+
         # If the parameter is a path parameter, we need to set the in_ field to "path".
         is_path_param = param_name in path_param_names
 

aws_lambda_powertools/event_handler/request.py

@@ -0,0 +1,115 @@
+"""Resolved HTTP Request object for Event Handler."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
+
+
+class Request:
+    """Represents the resolved HTTP request.
+
+    Provides structured access to the matched route pattern, extracted path parameters,
+    HTTP method, headers, query parameters, and body.  Available via ``app.request``
+    inside middleware and, when added as a type-annotated parameter, inside route handlers.
+
+    Examples
+    --------
+    **Middleware usage**
+
+    ```python
+    from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Request, Response
+    from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
+
+    app = APIGatewayRestResolver()
+
+    def auth_middleware(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
+        request: Request = app.request
+
+        route = request.route              # "/applications/{application_id}"
+        path_params = request.path_parameters  # {"application_id": "4da715ee-..."}
+        method = request.method            # "PUT"
+
+        if not is_authorized(route, method, path_params):
+            return Response(status_code=403, body="Forbidden")
+
+        return next_middleware(app)
+
+    app.use(middlewares=[auth_middleware])
+    ```
+
+    **Route handler injection (type-annotated)**
+
+    ```python
+    from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Request
+
+    app = APIGatewayRestResolver()
+
+    @app.get("/applications/<application_id>")
+    def get_application(application_id: str, request: Request):
+        user_agent = request.headers.get("user-agent")
+        return {"id": application_id, "user_agent": user_agent}
+    ```
+    """
+
+    __slots__ = ("_current_event", "_path_parameters", "_route_path")
+
+    def __init__(
+        self,
+        route_path: str,
+        path_parameters: dict[str, Any],
+        current_event: BaseProxyEvent,
+    ) -> None:
+        self._route_path = route_path
+        self._path_parameters = path_parameters
+        self._current_event = current_event
+
+    @property
+    def route(self) -> str:
+        """Matched route pattern in OpenAPI path-template format.
+
+        Examples
+        --------
+        For a route registered as ``/applications/<application_id>`` the value is
+        ``/applications/{application_id}``.
+        """
+        return self._route_path
+
+    @property
+    def path_parameters(self) -> dict[str, Any]:
+        """Extracted path parameters for the matched route.
+
+        Examples
+        --------
+        For a request to ``/applications/4da715ee``, matched against
+        ``/applications/<application_id>``, the value is
+        ``{"application_id": "4da715ee"}``.
+        """
+        return self._path_parameters
+
+    @property
+    def method(self) -> str:
+        """HTTP method in upper-case, e.g. ``"GET"``, ``"PUT"``."""
+        return self._current_event.http_method.upper()
+
+    @property
+    def headers(self) -> dict[str, str]:
+        """Request headers dict (lower-cased keys may vary by event source)."""
+        return self._current_event.headers or {}
+
+    @property
+    def query_parameters(self) -> dict[str, str] | None:
+        """Query string parameters, or ``None`` when none are present."""
+        return self._current_event.query_string_parameters
+
+    @property
+    def body(self) -> str | None:
+        """Raw request body string, or ``None`` when the request has no body."""
+        return self._current_event.body
+
+    @property
+    def json_body(self) -> Any:
+        """Request body deserialized as a Python object (dict / list), or ``None``."""
+        return self._current_event.json_body