feat(event_handler): support Depends() injection in function-based middleware

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1467,7 +1467,8 @@ def __call__(self, app: ApiGatewayResolver) -> dict | tuple | Response:
         logger.debug("MiddlewareFrame: %s", self)
         app._push_processed_stack_frame(str(self))
 
-        return self.current_middleware(app, self.next_middleware)
+        extra_kwargs = _resolve_extra_middleware_kwargs(app, self.current_middleware)
+        return self.current_middleware(app, self.next_middleware, **extra_kwargs)
 
 
 def _find_request_param_name(func: Callable) -> str | None:
@@ -1488,6 +1489,61 @@ def _find_request_param_name(func: Callable) -> str | None:
     return None
 
 
+_DEPENDS_CACHE_KEY = "_dependency_cache"
+
+
+def _inspect_middleware(middleware: Callable[..., Any]) -> tuple:
+    """Inspect a middleware's signature once and cache the result on the callable.
+
+    Returns a tuple of the ``Request``-typed parameter name (or ``None``) and the
+    middleware's dependency tree. Results are cached on the callable itself, mirroring
+    how ``Route`` caches its dependant, to avoid repeated signature analysis per request.
+    """
+    from aws_lambda_powertools.event_handler.depends import build_dependency_tree
+
+    cached = getattr(middleware, "_powertools_middleware_inspection", None)
+    if cached is not None:
+        return cached
+
+    inspection = (_find_request_param_name(middleware), build_dependency_tree(middleware))
+    try:
+        middleware._powertools_middleware_inspection = inspection  # type: ignore[attr-defined]
+    except (AttributeError, TypeError):
+        pass  # callables without attribute support are re-inspected per request
+
+    return inspection
+
+
+def _resolve_extra_middleware_kwargs(app: ApiGatewayResolver, middleware: Callable[..., Any]) -> dict:
+    """Resolve extra keyword arguments for a function-based middleware.
+
+    Supports ``Request`` injection and ``Depends()`` parameters, sharing the per-request
+    dependency cache (stored in ``app.context``) with route handlers so the same
+    dependency is resolved only once across middleware and routes. Class-based
+    middleware is intentionally out of scope.
+    """
+    from aws_lambda_powertools.event_handler.depends import solve_dependencies
+
+    request_param, dependency_tree = _inspect_middleware(middleware)
+
+    kwargs: dict = {}
+    if request_param:
+        kwargs[request_param] = app.request
+
+    if dependency_tree.dependencies:
+        dependency_cache = app.context.setdefault(_DEPENDS_CACHE_KEY, {})
+        kwargs.update(
+            solve_dependencies(
+                dependant=dependency_tree,
+                request=app.request,
+                dependency_overrides=app.dependency_overrides or None,
+                dependency_cache=dependency_cache,
+            ),
+        )
+
+    return kwargs
+
+
 def _registered_api_adapter(
     app: ApiGatewayResolver,
     next_middleware: Callable[..., Any],
@@ -1535,6 +1591,7 @@ def _registered_api_adapter(
                 dependant=build_dependency_tree(route.func),
                 request=app.request,
                 dependency_overrides=app.dependency_overrides or None,
+                dependency_cache=app.context.get(_DEPENDS_CACHE_KEY),
             )
             route_args.update(dep_values)
 

aws_lambda_powertools/event_handler/middlewares/async_utils.py

@@ -155,16 +155,20 @@ async def __call__(self, app: ApiGatewayResolver) -> dict | tuple | Response:
         logger.debug("AsyncMiddlewareFrame: %s", self)
         app._push_processed_stack_frame(str(self))
 
+        from aws_lambda_powertools.event_handler.api_gateway import _resolve_extra_middleware_kwargs
+
+        extra_kwargs = _resolve_extra_middleware_kwargs(app, self.current_middleware)
+
         if inspect.iscoroutinefunction(self.current_middleware):
-            return await self.current_middleware(app, self.next_middleware)
+            return await self.current_middleware(app, self.next_middleware, **extra_kwargs)
 
         loop = asyncio.get_running_loop()
 
         def sync_next(app: ApiGatewayResolver) -> Any:
             future = asyncio.run_coroutine_threadsafe(self.next_middleware(app), loop)
             return future.result()
 
-        return await asyncio.to_thread(self.current_middleware, app, sync_next)
+        return await asyncio.to_thread(self.current_middleware, app, sync_next, **extra_kwargs)
 
 
 async def _registered_api_adapter_async(
@@ -207,12 +211,14 @@ async def _registered_api_adapter_async(
             route_args = {**route_args, route.request_param_name: app.request}
 
         if route.has_dependencies:
+            from aws_lambda_powertools.event_handler.api_gateway import _DEPENDS_CACHE_KEY
             from aws_lambda_powertools.event_handler.depends import build_dependency_tree, solve_dependencies
 
             dep_values = solve_dependencies(
                 dependant=build_dependency_tree(route.func),
                 request=app.request,
                 dependency_overrides=app.dependency_overrides or None,
+                dependency_cache=app.context.get(_DEPENDS_CACHE_KEY),
             )
             route_args.update(dep_values)
 

docs/core/event_handler/api_gateway.md

@@ -1457,6 +1457,14 @@ The bridge between them is `request.context`: middleware writes to `app.context`
 --8<-- "examples/event_handler_rest/src/dependency_injection_with_middleware.py"
 ```
 
+Middleware can also declare `Depends()` parameters directly, the same way route handlers do. This is useful when the middleware itself needs typed data access - for example, looking up a user via a DAO before deciding whether to continue:
+
+```python hl_lines="22 25"
+--8<-- "examples/event_handler_rest/src/dependency_injection_middleware_depends.py"
+```
+
+Dependencies declared by middleware share the per-invocation cache with route dependencies, support `dependency_overrides` in tests, and can declare a `Request` parameter. This works for global (`app.use()`), router-level, and route-level (`@app.get(..., middlewares=[...])`) middleware.
+
 ???+ tip "When to use middleware vs Depends()"
     | Use case | Middleware | Depends() |
     | --- | --- | --- |

examples/event_handler_rest/src/dependency_injection_middleware_depends.py

@@ -0,0 +1,46 @@
+from typing import Annotated
+
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+from aws_lambda_powertools.event_handler.depends import Depends
+from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+app = APIGatewayHttpResolver()
+
+
+class UserDao:
+    def get(self, sub: str) -> dict | None:
+        # fetch user from a database
+        ...
+
+
+def get_user_dao() -> UserDao:
+    return UserDao()
+
+
+# Middleware can declare Depends() parameters, just like route handlers
+def user_middleware(
+    app: APIGatewayHttpResolver,
+    next_middleware: NextMiddleware,
+    user_dao: Annotated[UserDao, Depends(get_user_dao)],
+):
+    authorizer = app.current_event.request_context.authorizer
+    if authorizer:
+        sub: str | None = authorizer.jwt_claim.get("sub")
+        if sub:
+            user = user_dao.get(sub)
+            if user:
+                app.append_context(user=user)
+    return next_middleware(app)
+
+
+app.use(middlewares=[user_middleware])
+
+
+@app.get("/profile")
+def get_profile():
+    return {"user": app.context.get("user")}
+
+
+def lambda_handler(event: dict, context: LambdaContext) -> dict:
+    return app.resolve(event, context)