Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion aws_lambda_powertools/event_handler/api_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -1482,8 +1482,11 @@ def _find_request_param_name(func: Callable) -> str | None:
except Exception:
hints = {}

from aws_lambda_powertools.event_handler.depends import _unwrap_optional

for param_name, annotation in hints.items():
if annotation is Request:
# Python 3.10 wraps annotations of ``= None`` parameters in Optional[...]
if _unwrap_optional(annotation) is Request:
return param_name

return None
Expand Down
20 changes: 18 additions & 2 deletions aws_lambda_powertools/event_handler/depends.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@

from __future__ import annotations

from typing import TYPE_CHECKING, Annotated, Any, get_args, get_origin, get_type_hints
from types import UnionType
from typing import TYPE_CHECKING, Annotated, Any, Union, get_args, get_origin, get_type_hints

if TYPE_CHECKING:
from collections.abc import Callable
Expand Down Expand Up @@ -89,8 +90,23 @@ def __init__(self, *, param_name: str, depends: Depends, dependant: Dependant) -
self.dependant = dependant


def _unwrap_optional(annotation: Any) -> Any:
"""Return the inner type of ``Optional[X]`` / ``X | None``, or the annotation itself.
On Python 3.10, ``typing.get_type_hints`` wraps the annotation of a parameter whose
default is ``None`` in ``Optional[...]``, which would hide ``Annotated`` metadata such
as ``Depends()`` or a ``Request`` annotation. Python 3.11 stopped doing that.
"""
if get_origin(annotation) in (Union, UnionType):
args = [arg for arg in get_args(annotation) if arg is not type(None)]
if len(args) == 1:
return args[0]
return annotation


def _get_depends_from_annotation(annotation: Any) -> Depends | None:
"""Extract a Depends instance from an Annotated[Type, Depends(...)] annotation."""
annotation = _unwrap_optional(annotation)
if get_origin(annotation) is Annotated:
for arg in get_args(annotation)[1:]:
if isinstance(arg, Depends):
Expand Down Expand Up @@ -202,7 +218,7 @@ def solve_dependencies(
except Exception: # pragma: no cover - defensive for broken annotations
hints = {}
for param_name, annotation in hints.items():
if annotation is RequestClass:
if _unwrap_optional(annotation) is RequestClass:
sub_values[param_name] = request

try:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -645,3 +645,56 @@ def handler(user: Annotated[str, Depends(get_current_user)]):
result = app.resolve(load_event(event_file), {})
assert result["statusCode"] == 401
assert json.loads(result["body"]) == {"statusCode": 401, "message": "Missing authentication"}


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()

def get_region() -> str:
return "eu-west-1"

@app.post("/my/path")
def handler(region: Annotated[str, Depends(get_region)] = None):
return {"region": region}

result = app(API_GW_V2_EVENT, {})
assert result["statusCode"] == 200
assert json.loads(result["body"]) == {"region": "eu-west-1"}


def test_depends_inside_explicit_optional_is_resolved():
"""``Optional[Annotated[str, Depends(...)]]`` and ``Annotated[...] | None`` both resolve."""
from typing import Optional

app = APIGatewayHttpResolver()

def get_region() -> str:
return "eu-west-1"

def get_zone() -> str:
return "a"

@app.post("/my/path")
def handler(
region: Optional[Annotated[str, Depends(get_region)]] = None, # noqa: UP045
zone: Annotated[str, Depends(get_zone)] | None = None,
):
return {"region": region, "zone": zone}

result = app(API_GW_V2_EVENT, {})
assert result["statusCode"] == 200
assert json.loads(result["body"]) == {"region": "eu-west-1", "zone": "a"}


def test_request_param_with_none_default_is_injected():
"""A ``request: Request = None`` parameter is injected on every Python version."""
app = APIGatewayHttpResolver()

@app.post("/my/path")
def handler(request: Request = None):
return {"method": request.method}

result = app(API_GW_V2_EVENT, {})
assert result["statusCode"] == 200
assert json.loads(result["body"]) == {"method": "POST"}
Loading