feat: add Dependency injection feature

Author: leandrodamascena

aws_lambda_powertools/event_handler/api_gateway.py

@@ -472,6 +472,9 @@ def __init__(
 
         self.custom_response_validation_http_code = custom_response_validation_http_code
 
+        # Cache whether this route's handler declares Depends() parameters
+        self._has_dependencies: bool | None = None
+
         # 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
@@ -613,6 +616,21 @@ def dependant(self) -> Dependant:
 
         return self._dependant
 
+    @property
+    def has_dependencies(self) -> bool:
+        """Check if handler declares Depends() parameters without triggering full dependant computation."""
+        if self._has_dependencies is None:
+            from aws_lambda_powertools.event_handler.openapi.dependant import (
+                _get_depends_from_annotation,
+                get_typed_signature,
+            )
+
+            sig = get_typed_signature(self.func)
+            self._has_dependencies = any(
+                _get_depends_from_annotation(p.annotation) is not None for p in sig.parameters.values()
+            )
+        return self._has_dependencies
+
     @property
     def body_field(self) -> ModelField | None:
         if self._body_field is None:
@@ -1428,6 +1446,17 @@ def _registered_api_adapter(
         if route.request_param_name:
             route_args = {**route_args, route.request_param_name: app.request}
 
+        # Resolve Depends() parameters
+        if route.has_dependencies:
+            from aws_lambda_powertools.event_handler.openapi.dependant import solve_dependencies
+
+            dep_values = solve_dependencies(
+                dependant=route.dependant,
+                request=app.request,
+                dependency_overrides=app.dependency_overrides or None,
+            )
+            route_args.update(dep_values)
+
     return app._to_response(next_middleware(**route_args))
 
 
@@ -1496,6 +1525,7 @@ def __init__(
             by default json.loads when integrating with EventSource data class
         """
         self._proxy_type = proxy_type
+        self.dependency_overrides: dict[Callable, Callable] = {}
         self._dynamic_routes: list[Route] = []
         self._static_routes: list[Route] = []
         self._route_keys: list[str] = []

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -2,7 +2,9 @@
 
 import inspect
 import re
-from typing import TYPE_CHECKING, Any, ForwardRef, cast
+from typing import TYPE_CHECKING, Any, ForwardRef, cast, get_type_hints
+
+from typing_extensions import Annotated, get_args, get_origin
 
 from aws_lambda_powertools.event_handler.openapi.compat import (
     ModelField,
@@ -13,6 +15,8 @@
 from aws_lambda_powertools.event_handler.openapi.params import (
     Body,
     Dependant,
+    DependencyParam,
+    Depends,
     File,
     Form,
     Param,
@@ -149,6 +153,15 @@ def get_path_param_names(path: str) -> set[str]:
     return set(re.findall("{(.*?)}", path))
 
 
+def _get_depends_from_annotation(annotation: Any) -> Depends | None:
+    """Extract a Depends instance from an Annotated[Type, Depends(...)] annotation."""
+    if get_origin(annotation) is Annotated:
+        for arg in get_args(annotation)[1:]:
+            if isinstance(arg, Depends):
+                return arg
+    return None
+
+
 def get_dependant(
     *,
     path: str,
@@ -193,6 +206,22 @@ def get_dependant(
         if param.annotation is Request:
             continue
 
+        # Depends() parameters (via Annotated[Type, Depends(fn)]) are resolved at call time.
+        depends_instance = _get_depends_from_annotation(param.annotation)
+        if depends_instance is not None:
+            sub_dependant = get_dependant(
+                path=path,
+                call=depends_instance.dependency,
+            )
+            dependant.dependencies.append(
+                DependencyParam(
+                    param_name=param_name,
+                    depends=depends_instance,
+                    dependant=sub_dependant,
+                ),
+            )
+            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
 
@@ -386,3 +415,75 @@ def get_body_field_info(
             body_field_info_kwargs["media_type"] = body_param_media_types[0]
 
     return body_field_info, body_field_info_kwargs
+
+
+def solve_dependencies(
+    *,
+    dependant: Dependant,
+    request: Request | None = None,
+    dependency_overrides: dict[Callable[..., Any], Callable[..., Any]] | None = None,
+    dependency_cache: dict[Callable[..., Any], Any] | None = None,
+) -> dict[str, Any]:
+    """
+    Recursively resolve all ``Depends()`` parameters for a given dependant.
+
+    Parameters
+    ----------
+    dependant: Dependant
+        The dependant model containing dependency declarations
+    request: Request, optional
+        The current request object, injected into dependencies that declare a Request parameter
+    dependency_overrides: dict, optional
+        Mapping of original dependency callable to override callable (for testing)
+    dependency_cache: dict, optional
+        Per-invocation cache of resolved dependency values
+
+    Returns
+    -------
+    dict[str, Any]
+        Mapping of parameter name to resolved dependency value
+    """
+    if dependency_cache is None:
+        dependency_cache = {}
+
+    values: dict[str, Any] = {}
+
+    for dep in dependant.dependencies:
+        use_fn = dep.depends.dependency
+
+        # Apply overrides (for testing)
+        if dependency_overrides and use_fn in dependency_overrides:
+            use_fn = dependency_overrides[use_fn]
+
+        # Check cache
+        if dep.depends.use_cache and use_fn in dependency_cache:
+            values[dep.param_name] = dependency_cache[use_fn]
+            continue
+
+        # Recursively resolve sub-dependencies
+        sub_values = solve_dependencies(
+            dependant=dep.dependant,
+            request=request,
+            dependency_overrides=dependency_overrides,
+            dependency_cache=dependency_cache,
+        )
+
+        # Inject Request if the dependency declares it
+        if request is not None:
+            try:
+                hints = get_type_hints(use_fn)
+            except Exception:
+                hints = {}
+            for param_name, annotation in hints.items():
+                if annotation is Request:
+                    sub_values[param_name] = request
+
+        solved = use_fn(**sub_values)
+
+        # Cache result
+        if dep.depends.use_cache:
+            dependency_cache[use_fn] = solved
+
+        values[dep.param_name] = solved
+
+    return values

aws_lambda_powertools/event_handler/openapi/params.py

@@ -42,6 +42,47 @@ class ParamTypes(Enum):
 _Unset: Any = Undefined
 
 
+class Depends:
+    """
+    Declares a dependency for a route handler parameter.
+
+    Dependencies are resolved automatically before the handler is called. The return value
+    of the dependency callable is injected as the parameter value.
+
+    Parameters
+    ----------
+    dependency: Callable[..., Any]
+        A callable whose return value will be injected into the handler parameter.
+        The callable can itself declare ``Depends()`` parameters to form a dependency tree.
+    use_cache: bool
+        If ``True`` (default), the dependency result is cached per invocation so that
+        the same dependency used multiple times is only called once.
+
+    Examples
+    --------
+
+    ```python
+    from typing_extensions import Annotated
+
+    from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+    from aws_lambda_powertools.event_handler.openapi.params import Depends
+
+    app = APIGatewayHttpResolver()
+
+    def get_tenant() -> str:
+        return "default-tenant"
+
+    @app.get("/orders")
+    def list_orders(tenant_id: Annotated[str, Depends(get_tenant)]):
+        return {"tenant": tenant_id}
+    ```
+    """
+
+    def __init__(self, dependency: Callable[..., Any], *, use_cache: bool = True) -> None:
+        self.dependency = dependency
+        self.use_cache = use_cache
+
+
 class Dependant:
     """
     A class used internally to represent a dependency between path operation decorators and the path operation function.
@@ -64,6 +105,7 @@ def __init__(
         http_connection_param_name: str | None = None,
         response_param_name: str | None = None,
         background_tasks_param_name: str | None = None,
+        dependencies: list[DependencyParam] | None = None,
         path: str | None = None,
     ) -> None:
         self.path_params = path_params or []
@@ -78,6 +120,7 @@ def __init__(
         self.http_connection_param_name = http_connection_param_name
         self.response_param_name = response_param_name
         self.background_tasks_param_name = background_tasks_param_name
+        self.dependencies = dependencies or []
         self.name = name
         self.call = call
         # Store the path to be able to re-generate a dependable from it in overrides
@@ -86,6 +129,15 @@ def __init__(
         self.cache_key: CacheKey = self.call
 
 
+class DependencyParam:
+    """Holds a dependency's parameter name and its resolved Dependant sub-tree."""
+
+    def __init__(self, *, param_name: str, depends: Depends, dependant: Dependant) -> None:
+        self.param_name = param_name
+        self.depends = depends
+        self.dependant = dependant
+
+
 class Param(FieldInfo):  # type: ignore[misc]
     """
     A class used internally to represent a parameter in a path operation.
@@ -816,7 +868,7 @@ def get_flat_dependant(
         visited = []
     visited.append(dependant.cache_key)
 
-    return Dependant(
+    flat = Dependant(
         path_params=dependant.path_params.copy(),
         query_params=dependant.query_params.copy(),
         header_params=dependant.header_params.copy(),
@@ -825,6 +877,18 @@ def get_flat_dependant(
         path=dependant.path,
     )
 
+    # Flatten sub-dependencies that declare HTTP params (query, header, etc.)
+    for dep in dependant.dependencies:
+        if dep.dependant.cache_key not in visited:
+            sub_flat = get_flat_dependant(dep.dependant, visited=visited)
+            flat.path_params.extend(sub_flat.path_params)
+            flat.query_params.extend(sub_flat.query_params)
+            flat.header_params.extend(sub_flat.header_params)
+            flat.cookie_params.extend(sub_flat.cookie_params)
+            flat.body_params.extend(sub_flat.body_params)
+
+    return flat
+
 
 def analyze_param(
     *,

docs/core/event_handler/api_gateway.md

@@ -1365,6 +1365,48 @@ You can use `append_context` when you want to share data between your App and Ro
     --8<-- "examples/event_handler_rest/src/split_route_append_context_module.py"
 	```
 
+### Dependency injection
+
+You can use `Depends()` to declare dependencies that are automatically resolved and injected into your route handlers. This provides type-safe, composable, and testable dependency injection.
+
+#### Basic usage
+
+Use `Annotated[Type, Depends(fn)]` to declare a dependency. The return value of `fn` is injected into the parameter automatically.
+
+```python hl_lines="5 8 20 25"
+--8<-- "examples/event_handler_rest/src/dependency_injection.py"
+```
+
+#### Nested dependencies
+
+Dependencies can depend on other dependencies, forming a composable tree. Shared sub-dependencies are resolved once per invocation and cached automatically.
+
+```python hl_lines="18 22 29-30"
+--8<-- "examples/event_handler_rest/src/dependency_injection_nested.py"
+```
+
+#### Accessing the request
+
+Dependencies that need access to the current request can declare a parameter typed as `Request`. It will be injected automatically.
+
+```python hl_lines="5-6 12 20"
+--8<-- "examples/event_handler_rest/src/dependency_injection_with_request.py"
+```
+
+#### Testing with dependency overrides
+
+Use `dependency_overrides` to replace any dependency with a mock or stub during testing - no monkeypatching needed.
+
+```python hl_lines="3 12 26"
+--8<-- "examples/event_handler_rest/src/dependency_injection_testing.py"
+```
+
+???+ tip "Caching behavior"
+    By default, dependencies are cached within the same invocation (`use_cache=True`). If the same dependency is used by multiple handlers or sub-dependencies, it is resolved once and the result is reused. Use `Depends(fn, use_cache=False)` to resolve every time.
+
+???+ info "`append_context` vs `Depends()`"
+    `append_context` remains available for backward compatibility. `Depends()` is recommended for new code because it provides type safety, IDE autocomplete, composable dependency trees, and `dependency_overrides` for testing.
+
 #### Sample layout
 
 This is a sample project layout for a monolithic function with routes split in different files (`/todos`, `/health`).

examples/event_handler_rest/src/dependency_injection.py

@@ -0,0 +1,32 @@
+import os
+from typing import Any
+
+import boto3
+from typing_extensions import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+from aws_lambda_powertools.event_handler.openapi.params import Depends
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+app = APIGatewayHttpResolver()
+
+
+def get_dynamodb_table():
+    dynamodb = boto3.resource("dynamodb")
+    return dynamodb.Table(os.environ["TABLE_NAME"])
+
+
+@app.get("/orders")
+def list_orders(table: Annotated[Any, Depends(get_dynamodb_table)]):
+    return table.scan()["Items"]
+
+
+@app.post("/orders")
+def create_order(table: Annotated[Any, Depends(get_dynamodb_table)]):
+    order = app.current_event.json_body
+    table.put_item(Item=order)
+    return {"message": "Order created"}
+
+
+def lambda_handler(event: dict, context: LambdaContext) -> dict:
+    return app.resolve(event, context)