From 2179577730096e70fed4af01f715ad42f6f2f16f Mon Sep 17 00:00:00 2001 From: oyiz-michael Date: Mon, 16 Mar 2026 03:04:30 +0000 Subject: [PATCH] feat(event_handler): add Request object for middleware access to resolved route and args Introduce a Request class that provides structured access to the resolved route pattern, path parameters, HTTP method, headers, query parameters, and body. Available via app.request in middleware and via type-annotation injection in route handlers. Closes #7992, #4609 --- .../event_handler/__init__.py | 2 + .../event_handler/api_gateway.py | 69 +++ .../event_handler/openapi/dependant.py | 6 + .../event_handler/request.py | 115 ++++ .../required_dependencies/test_request.py | 513 ++++++++++++++++++ 5 files changed, 705 insertions(+) create mode 100644 aws_lambda_powertools/event_handler/request.py create mode 100644 tests/functional/event_handler/required_dependencies/test_request.py diff --git a/aws_lambda_powertools/event_handler/__init__.py b/aws_lambda_powertools/event_handler/__init__.py index 6b926e6248a..582abd017c0 100644 --- a/aws_lambda_powertools/event_handler/__init__.py +++ b/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", diff --git a/aws_lambda_powertools/event_handler/api_gateway.py b/aws_lambda_powertools/event_handler/api_gateway.py index b1e0c9ff16d..08e75c95204 100644 --- a/aws_lambda_powertools/event_handler/api_gateway.py +++ b/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 `. + + 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)) diff --git a/aws_lambda_powertools/event_handler/openapi/dependant.py b/aws_lambda_powertools/event_handler/openapi/dependant.py index 310cab68e66..db8fa0ad251 100644 --- a/aws_lambda_powertools/event_handler/openapi/dependant.py +++ b/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 diff --git a/aws_lambda_powertools/event_handler/request.py b/aws_lambda_powertools/event_handler/request.py new file mode 100644 index 00000000000..e402c094ded --- /dev/null +++ b/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/") + 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/`` 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/``, 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 diff --git a/tests/functional/event_handler/required_dependencies/test_request.py b/tests/functional/event_handler/required_dependencies/test_request.py new file mode 100644 index 00000000000..a8fb52fef68 --- /dev/null +++ b/tests/functional/event_handler/required_dependencies/test_request.py @@ -0,0 +1,513 @@ +"""Tests for the Request object feature (GH #7992). + +Covers: +- ``app.request`` availability in global and route-level middleware +- ``Request`` type-annotation injection in route handlers +- ``Request`` properties: route, path_parameters, method, headers, query_parameters, body +- ``RuntimeError`` when ``app.request`` is accessed outside of resolution +- Backward compatibility: routes without ``Request`` continue to work unchanged +- ``APIGatewayHttpResolver`` and ``ALBResolver`` variants +""" +from __future__ import annotations + +import pytest + +from aws_lambda_powertools.event_handler import ( + ALBResolver, + APIGatewayHttpResolver, + APIGatewayRestResolver, + Request, + Response, +) +from aws_lambda_powertools.event_handler.api_gateway import ProxyEventType +from aws_lambda_powertools.event_handler.middlewares import NextMiddleware +from tests.functional.utils import load_event + +# --------------------------------------------------------------------------- +# Shared test events +# --------------------------------------------------------------------------- + +API_REST_EVENT = load_event("apiGatewayProxyEvent.json") # GET /my/path +API_RESTV2_EVENT = load_event("apiGatewayProxyV2Event_GET.json") + + +def _make_rest_event(path: str, method: str = "GET", path_parameters: dict | None = None, body: str | None = None): + """Build a minimal API Gateway REST (v1) proxy event.""" + return { + "httpMethod": method, + "path": path, + "pathParameters": path_parameters, + "queryStringParameters": None, + "multiValueQueryStringParameters": None, + "headers": {"Content-Type": "application/json", "user-agent": "pytest"}, + "multiValueHeaders": {}, + "body": body, + "isBase64Encoded": False, + "requestContext": {"httpMethod": method, "resourcePath": path}, + "resource": path, + "stageVariables": None, + } + + +# --------------------------------------------------------------------------- +# app.request in global middleware +# --------------------------------------------------------------------------- + + +def test_request_available_in_global_middleware(): + app = APIGatewayRestResolver() + captured: list[Request] = [] + + def capture_middleware(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[capture_middleware]) + + @app.get("/my/path") + def handler(): + return {} + + app(API_REST_EVENT, {}) + + assert len(captured) == 1 + req = captured[0] + assert isinstance(req, Request) + assert req.route == "/my/path" + assert req.method == "GET" + + +def test_request_route_pattern_uses_openapi_format(): + """route property should use {param} OpenAPI notation, not Powertools notation.""" + app = APIGatewayRestResolver() + captured: list[Request] = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.get("/applications/") + def handler(application_id: str): + return {} + + event = _make_rest_event( + "/applications/42", + path_parameters={"application_id": "42"}, + ) + app(event, {}) + + assert captured[0].route == "/applications/{application_id}" + + +def test_request_path_parameters_in_middleware(): + app = APIGatewayRestResolver() + captured: list[dict] = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + captured.append(app.request.path_parameters) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.get("/applications/") + def handler(application_id: str): + return {} + + event = _make_rest_event( + "/applications/4da715ee", + path_parameters={"application_id": "4da715ee"}, + ) + app(event, {}) + + assert captured == [{"application_id": "4da715ee"}] + + +def test_request_method_in_middleware(): + app = APIGatewayRestResolver() + methods_seen: list[str] = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + methods_seen.append(app.request.method) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.put("/items/") + def handler(item_id: str): + return {} + + event = _make_rest_event("/items/99", method="PUT", path_parameters={"item_id": "99"}) + app(event, {}) + + assert methods_seen == ["PUT"] + + +def test_request_headers_in_middleware(): + app = APIGatewayRestResolver() + headers_seen: list[dict] = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + headers_seen.append(app.request.headers) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.get("/my/path") + def handler(): + return {} + + app(API_REST_EVENT, {}) + + assert len(headers_seen) == 1 + # headers is a dict (may have varying casing depending on event source) + assert isinstance(headers_seen[0], dict) + + +def test_request_query_parameters_in_middleware(): + app = APIGatewayRestResolver() + captured: list = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + captured.append(app.request.query_parameters) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.get("/search") + def handler(): + return {} + + event = _make_rest_event("/search") + event["queryStringParameters"] = {"q": "powertools"} + app(event, {}) + + assert captured == [{"q": "powertools"}] + + +def test_request_body_in_middleware(): + app = APIGatewayRestResolver() + bodies_seen: list = [] + + def mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + bodies_seen.append(app.request.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"}') + event["httpMethod"] = "POST" + app(event, {}) + + assert bodies_seen == ['{"name": "widget"}'] + + +# --------------------------------------------------------------------------- +# Request injection in route handlers via type annotation +# --------------------------------------------------------------------------- + + +def test_request_injected_into_handler(): + app = APIGatewayRestResolver() + + received: list[Request] = [] + + @app.get("/my/path") + def handler(request: Request): + received.append(request) + return {} + + app(API_REST_EVENT, {}) + + assert len(received) == 1 + assert isinstance(received[0], Request) + assert received[0].route == "/my/path" + assert received[0].method == "GET" + + +def test_request_injected_alongside_path_params(): + app = APIGatewayRestResolver() + + received: list[tuple] = [] + + @app.get("/users/") + def handler(user_id: str, request: Request): + received.append((user_id, request)) + return {} + + event = _make_rest_event("/users/123", path_parameters={"user_id": "123"}) + app(event, {}) + + assert len(received) == 1 + user_id, req = received[0] + assert user_id == "123" + assert isinstance(req, Request) + assert req.path_parameters == {"user_id": "123"} + assert req.route == "/users/{user_id}" + + +def test_request_injection_parameter_name_is_flexible(): + """The parameter can be named anything as long as it is annotated as Request.""" + app = APIGatewayRestResolver() + + received: list[Request] = [] + + @app.get("/my/path") + def handler(req: Request): + received.append(req) + return {} + + app(API_REST_EVENT, {}) + + assert received[0].route == "/my/path" + + +def test_handler_without_request_annotation_unaffected(): + """Existing handlers with no Request annotation continue to work identically.""" + app = APIGatewayRestResolver() + + @app.get("/my/path") + def handler(): + return {"ok": True} + + result = app(API_REST_EVENT, {}) + assert result["statusCode"] == 200 + + +def test_handler_with_path_params_only_unaffected(): + """Handlers that only use path params continue to work identically.""" + app = APIGatewayRestResolver() + + @app.get("/users/") + def handler(user_id: str): + return {"id": user_id} + + event = _make_rest_event("/users/42", path_parameters={"user_id": "42"}) + result = app(event, {}) + assert result["statusCode"] == 200 + + +# --------------------------------------------------------------------------- +# Request injection caching (idempotency across multiple calls) +# --------------------------------------------------------------------------- + + +def test_request_injection_works_across_multiple_invocations(): + """Injection must work correctly on repeated calls (cached param name must stay valid).""" + app = APIGatewayRestResolver() + call_count = 0 + + @app.get("/counters/") + def handler(counter_id: str, request: Request): + nonlocal call_count + call_count += 1 + assert request.path_parameters["counter_id"] == counter_id + return {} + + for i in range(3): + event = _make_rest_event(f"/counters/{i}", path_parameters={"counter_id": str(i)}) + result = app(event, {}) + assert result["statusCode"] == 200 + + assert call_count == 3 + + +# --------------------------------------------------------------------------- +# RuntimeError when accessed outside of request resolution +# --------------------------------------------------------------------------- + + +def test_request_raises_before_resolution(): + app = APIGatewayRestResolver() + with pytest.raises(RuntimeError, match="app.request is only available after route resolution"): + _ = app.request + + +# --------------------------------------------------------------------------- +# Route-level middleware also gets app.request +# --------------------------------------------------------------------------- + + +def test_request_available_in_route_level_middleware(): + app = APIGatewayRestResolver() + captured: list[Request] = [] + + def route_mw(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response: + captured.append(app.request) + return next_middleware(app) + + @app.get("/protected/", middlewares=[route_mw]) + def handler(resource_id: str): + return {} + + event = _make_rest_event("/protected/abc", path_parameters={"resource_id": "abc"}) + app(event, {}) + + assert len(captured) == 1 + assert captured[0].route == "/protected/{resource_id}" + assert captured[0].path_parameters == {"resource_id": "abc"} + + +# --------------------------------------------------------------------------- +# Other resolver types +# --------------------------------------------------------------------------- + + +def test_request_available_in_http_resolver_middleware(): + app = APIGatewayHttpResolver() + captured: list[Request] = [] + + def mw(app, next_middleware): + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @app.get("/my/path") + def handler(): + return {} + + app(API_RESTV2_EVENT, {}) + + assert len(captured) == 1 + assert captured[0].method == "GET" + + +def test_request_available_in_alb_middleware(): + alb_event = load_event("albEvent.json") + app = ALBResolver() + captured: list[Request] = [] + + def mw(app, next_middleware): + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[mw]) + + # Register a route that matches the ALB event's path + path = alb_event.get("path", "/lambda") + + @app.get(path) + def handler(): + return {} + + app(alb_event, {}) + + assert len(captured) == 1 + assert isinstance(captured[0], Request) + + +# --------------------------------------------------------------------------- +# Router / include_router pattern +# --------------------------------------------------------------------------- + + +def test_request_available_in_middleware_with_include_router(): + """app.request must work in middleware when routes come from an included Router.""" + from aws_lambda_powertools.event_handler.api_gateway import Router + + app = APIGatewayRestResolver() + router = Router() + captured: list[Request] = [] + + def mw(app, next_middleware): + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[mw]) + + @router.get("/users/") + def get_user(user_id: str): + return {"id": user_id} + + app.include_router(router) + + event = _make_rest_event("/users/abc", path_parameters={"user_id": "abc"}) + result = app(event, {}) + + assert result["statusCode"] == 200 + assert len(captured) == 1 + assert captured[0].route == "/users/{user_id}" + assert captured[0].path_parameters == {"user_id": "abc"} + + +def test_request_injected_in_handler_with_include_router(): + """Request injection via type annotation must work when routes come from an included Router.""" + from aws_lambda_powertools.event_handler.api_gateway import Router + + app = APIGatewayRestResolver() + router = Router() + received: list[Request] = [] + + @router.get("/items/") + def get_item(item_id: str, request: Request): + received.append(request) + return {"id": item_id} + + app.include_router(router) + + event = _make_rest_event("/items/xyz", path_parameters={"item_id": "xyz"}) + result = app(event, {}) + + assert result["statusCode"] == 200 + assert len(received) == 1 + assert received[0].route == "/items/{item_id}" + assert received[0].path_parameters == {"item_id": "xyz"} + + +# --------------------------------------------------------------------------- +# Proxy+ use case (the original issue scenario) +# --------------------------------------------------------------------------- + + +def test_request_resolves_path_params_from_proxy_plus_event(): + """When API GW uses {proxy+}, app.current_event.pathParameters only has 'proxy'. + But app.request.path_parameters should have the *resolved* params from Powertools routing.""" + app = APIGatewayRestResolver() + captured: list[Request] = [] + + def auth_middleware(app, next_middleware): + captured.append(app.request) + return next_middleware(app) + + app.use(middlewares=[auth_middleware]) + + @app.get("/applications/") + def get_application(application_id: str): + return {"id": application_id} + + @app.put("/applications/") + def put_application(application_id: str): + return {"updated": application_id} + + # Simulate a proxy+ event where API GW only knows about {proxy+} + event = { + "httpMethod": "PUT", + "path": "/applications/4da715ee-79d4-4e52-81cb-1ecc464708fb", + "pathParameters": {"proxy": "4da715ee-79d4-4e52-81cb-1ecc464708fb"}, + "queryStringParameters": None, + "multiValueQueryStringParameters": None, + "headers": {"Content-Type": "application/json"}, + "multiValueHeaders": {}, + "body": None, + "isBase64Encoded": False, + "requestContext": {"httpMethod": "PUT", "resourcePath": "/applications/{proxy+}"}, + "resource": "/applications/{proxy+}", + "stageVariables": None, + } + + result = app(event, {}) + + assert result["statusCode"] == 200 + assert len(captured) == 1 + + req = captured[0] + # Middleware sees the resolved route, NOT the proxy+ pattern + assert req.route == "/applications/{application_id}" + assert req.path_parameters == {"application_id": "4da715ee-79d4-4e52-81cb-1ecc464708fb"} + assert req.method == "PUT"