feat(event_handler): adding resolve async public API

aws_lambda_powertools/event_handler/api_gateway.py

@@ -663,7 +663,7 @@
 
         app.append_context(_route_args=route_arguments)
 
-        # Build async chain from inside-out (not cached — avoids state conflicts with sync cache)
+        # Build async chain from inside-out (not cached, avoids state conflicts with sync cache)
         next_handler: Callable = self.func
         for handler in reversed(all_middlewares):
             next_handler = AsyncMiddlewareFrame(current_middleware=handler, next_middleware=next_handler)
@@ -2558,7 +2558,7 @@
         # Debug print Processed Middlewares
         if self._debug:
             print("\nProcessed Middlewares:")
             print("======================")
             print("\n".join(self.processed_stack_frames))
             print("======================")
 
@@ -2566,6 +2566,66 @@
 
         return response
 
+    async def resolve_async(self, event: Mapping[str, Any], context: LambdaContext) -> dict[str, Any]:
+        """Async version of resolve() for native async handler support.
+
+        Use this method when your route handlers use async/await. The resolution
+        pipeline supports both sync and async handlers transparently.
+
+        Parameters
+        ----------
+        event: dict[str, Any]
+            Event
+        context: LambdaContext
+            Lambda context
+        Returns
+        -------
+        dict
+            Returns the dict response
+
+        Example
+        -------
+
+        ```python
+        import asyncio
+        from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+
+        app = APIGatewayHttpResolver()
+
+        @app.get("/async")
+        async def async_handler():
+            return {"message": "async works"}
+
+        def lambda_handler(event, context):
+            return asyncio.run(app.resolve_async(event, context))
+        ```
+        """
+        if isinstance(event, BaseProxyEvent):
+            warnings.warn(
+                "You don't need to serialize event to Event Source Data Class when using Event Handler; "
+                "see issue #1152",
+                stacklevel=2,
+            )
+            event = event.raw_event
+
+        if self._debug:
+            print(self._serializer(cast(dict, event)))
+
+        BaseRouter.current_event = self._to_proxy_event(cast(dict, event))
+        BaseRouter.lambda_context = context
+
+        response = (await self._resolve_async()).build(self.current_event, self._cors)
+
+        if self._debug:
+            print("\nProcessed Middlewares:")
+            print("======================")
+            print("\n".join(self.processed_stack_frames))
+            print("======================")
+
+        self.clear_context()
+
+        return response
+
     async def _resolve_async(self) -> ResponseBuilder:
         method = self.current_event.http_method.upper()
         path = self._remove_prefix(self.current_event.path)

docs/core/event_handler/api_gateway.md

@@ -11,6 +11,7 @@ Event handler for Amazon API Gateway REST and HTTP APIs, Application Load Balanc
 * Support for CORS, binary and Gzip compression, Decimals JSON encoding and bring your own JSON serializer
 * Built-in integration with [Event Source Data Classes utilities](../../utilities/data_classes.md){target="_blank"} for self-documented event schema
 * Works with micro function (one or a few routes) and monolithic functions (all routes)
+* Native async handler support with `resolve_async()` for non-blocking I/O
 * Support for Middleware
 * Support for OpenAPI schema generation
 * Support data validation for requests/responses
@@ -1464,6 +1465,99 @@ Use `dependency_overrides` to replace any dependency with a mock or stub during
 ???+ 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.
 
+### Async support
+
+Use `resolve_async()` to natively support async route handlers with `async/await`. This enables non-blocking I/O operations like concurrent HTTP calls, database queries, and parallel processing within your Lambda function.
+
+Both sync and async handlers can coexist in the same resolver. Async handlers are automatically detected and awaited.
+
+=== "Getting started"
+
+    ```python hl_lines="9 22" title="async_resolve_getting_started.py"
+    --8<-- "examples/event_handler_rest/src/async_resolve_getting_started.py"
+    ```
+
+    1. Define your route handler as `async def` to use `await`
+    2. Sync handlers continue to work as before, no changes needed
+    3. Use `resolve_async()` instead of `resolve()` and wrap with `asyncio.run()`
+
+=== "Concurrent I/O with gather"
+
+    ```python hl_lines="21-24" title="async_resolve_concurrent.py"
+    --8<-- "examples/event_handler_rest/src/async_resolve_concurrent.py"
+    ```
+
+    1. `asyncio.gather()` runs multiple I/O operations concurrently, reducing total latency
+
+=== "All resolvers"
+
+    ```python hl_lines="1 10-12" title="async_resolve_all_resolvers.py"
+    --8<-- "examples/event_handler_rest/src/async_resolve_all_resolvers.py"
+    ```
+
+    1. API Gateway REST API
+    2. API Gateway HTTP API
+    3. Application Load Balancer
+
+#### Middlewares
+
+Both sync and async middlewares work in the async chain. Sync middlewares are executed in a background thread so the event loop is never blocked.
+
+=== "Sync middleware"
+
+    ```python hl_lines="11 24" title="async_resolve_middleware.py"
+    --8<-- "examples/event_handler_rest/src/async_resolve_middleware.py"
+    ```
+
+    1. Sync middleware works as-is, no changes needed
+    2. Async handler is awaited natively in the async chain
+
+=== "Async middleware"
+
+    ```python hl_lines="11 16" title="async_resolve_async_middleware.py"
+    --8<-- "examples/event_handler_rest/src/async_resolve_async_middleware.py"
+    ```
+
+    1. Define your middleware as `async def` to use `await`
+    2. Use `await next_middleware(app)` instead of `next_middleware(app)`
+
+#### Async with data validation
+
+Data validation with Pydantic works with async handlers. Use `enable_validation=True` as you would with sync handlers.
+
+```python hl_lines="1 3 7"
+app = APIGatewayHttpResolver(enable_validation=True)
+
+@app.get("/todos/<todo_id>")
+async def get_todo(todo_id: int) -> dict:
+    return {"todo_id": todo_id}
+
+def lambda_handler(event, context):
+    return asyncio.run(app.resolve_async(event, context))
+```
+
+#### Operations that remain synchronous
+
+These operations run synchronously on the event loop. They are CPU-bound and complete in microseconds, so they do not benefit from async.
+
+| Operation | Why it stays synchronous |
+| ----------------------------- | --------------------------------------------------------------- |
+| **Route matching** | Regex matching and string comparison against registered routes |
+| **Event deserialization** | Converting the raw event dict into a proxy event data class |
+| **Response serialization** | JSON encoding, base64 encoding, header assembly |
+| **Response validation** | Pydantic model validation is CPU-bound |
+| **Request validation** | Pydantic model validation is CPU-bound |
+| **Compression** | Gzip compression of response body |
+| **CORS header injection** | Building Access-Control headers from config |
+| **Dependency resolution** | `Depends()` tree is resolved synchronously |
+
+#### Known limitations
+
+| Limitation | Detail |
+| ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------- |
+| **AWS X-Ray with `asyncio.gather`** | X-Ray SDK does not propagate trace context across `asyncio.gather` tasks. Use individual `await` calls if you need per-call tracing. |
+| **Sync middlewares use thread pool** | Sync middlewares run in the default `ThreadPoolExecutor`. Avoid long blocking I/O inside sync middlewares when using `resolve_async()`. |
+
 ### Considerations
 
 This utility is optimized for fast startup, minimal feature set, and to quickly on-board customers familiar with frameworks like Flask — it's not meant to be a fully fledged framework.
@@ -1546,6 +1640,20 @@ Each endpoint will be it's own Lambda function that is configured as a [Lambda i
 
 ## Testing your code
 
+### Testing async handlers
+
+You can test async handlers by calling `resolve_async()` with `asyncio.run()`.
+
+```python hl_lines="24 26" title="async_resolve_testing.py"
+--8<-- "examples/event_handler_rest/src/async_resolve_testing.py"
+```
+
+1. Import your app as usual
+2. Use `asyncio.run(app.resolve_async(...))` instead of `app.resolve(...)`
+3. Assert on the response dict as you would with sync handlers
+
+### Testing sync handlers
+
 You can test your routes by passing a proxy event request with required params.
 
 ???+ info

examples/event_handler_rest/src/async_resolve_all_resolvers.py

@@ -0,0 +1,31 @@
+import asyncio
+
+from aws_lambda_powertools.event_handler import (
+    ALBResolver,
+    APIGatewayHttpResolver,
+    APIGatewayRestResolver,
+)
+
+rest_app = APIGatewayRestResolver()  # (1)!
+http_app = APIGatewayHttpResolver()  # (2)!
+alb_app = ALBResolver()  # (3)!
+
+
+@rest_app.get("/hello")
+@http_app.get("/hello")
+@alb_app.get("/hello")
+async def hello():
+    await asyncio.sleep(0)
+    return {"message": "hello from async"}
+
+
+def rest_handler(event, context):
+    return asyncio.run(rest_app.resolve_async(event, context))
+
+
+def http_handler(event, context):
+    return asyncio.run(http_app.resolve_async(event, context))
+
+
+def alb_handler(event, context):
+    return asyncio.run(alb_app.resolve_async(event, context))

examples/event_handler_rest/src/async_resolve_async_middleware.py

@@ -0,0 +1,29 @@
+import asyncio
+from collections.abc import Callable
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
+
+app = APIGatewayRestResolver()
+logger = Logger()
+
+
+async def async_inject_correlation_id(app: APIGatewayRestResolver, next_middleware: Callable) -> Response:  # (1)!
+    request_id = app.current_event.request_context.request_id
+    app.append_context(correlation_id=request_id)
+    logger.set_correlation_id(request_id)
+
+    result = await next_middleware(app)  # (2)!
+
+    result.headers["x-correlation-id"] = request_id
+    return result
+
+
+@app.get("/todos", middlewares=[async_inject_correlation_id])
+async def get_todos():
+    await asyncio.sleep(0)
+    return {"todos": []}
+
+
+def lambda_handler(event, context):
+    return asyncio.run(app.resolve_async(event, context))

examples/event_handler_rest/src/async_resolve_concurrent.py

@@ -0,0 +1,28 @@
+import asyncio
+
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+
+app = APIGatewayHttpResolver()
+
+
+async def fetch_profile(user_id: str) -> dict:
+    await asyncio.sleep(0)  # simulate async I/O (e.g., DynamoDB, HTTP call)
+    return {"user_id": user_id, "name": "John"}
+
+
+async def fetch_orders(user_id: str) -> list:
+    await asyncio.sleep(0)
+    return [{"order_id": "123", "total": 99.99}]
+
+
+@app.get("/dashboard/<user_id>")
+async def get_dashboard(user_id: str):
+    profile, orders = await asyncio.gather(  # (1)!
+        fetch_profile(user_id),
+        fetch_orders(user_id),
+    )
+    return {"profile": profile, "orders": orders}
+
+
+def lambda_handler(event, context):
+    return asyncio.run(app.resolve_async(event, context))

examples/event_handler_rest/src/async_resolve_getting_started.py

@@ -0,0 +1,21 @@
+import asyncio
+
+from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
+
+app = APIGatewayHttpResolver()
+
+
+@app.get("/todos/<todo_id>")
+async def get_todo(todo_id: str):  # (1)!
+    # Async handlers can use await for non-blocking I/O
+    await asyncio.sleep(0)  # simulate async I/O
+    return {"todo_id": todo_id, "completed": False}
+
+
+@app.get("/health")
+def health():  # (2)!
+    return {"status": "ok"}
+
+
+def lambda_handler(event, context):
+    return asyncio.run(app.resolve_async(event, context))  # (3)!

examples/event_handler_rest/src/async_resolve_middleware.py

@@ -0,0 +1,29 @@
+import asyncio
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
+from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
+
+app = APIGatewayRestResolver()
+logger = Logger()
+
+
+def inject_correlation_id(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:  # (1)!
+    request_id = app.current_event.request_context.request_id
+    app.append_context(correlation_id=request_id)
+    logger.set_correlation_id(request_id)
+
+    result = next_middleware(app)
+
+    result.headers["x-correlation-id"] = request_id
+    return result
+
+
+@app.get("/todos", middlewares=[inject_correlation_id])
+async def get_todos():  # (2)!
+    await asyncio.sleep(0)
+    return {"todos": []}
+
+
+def lambda_handler(event, context):
+    return asyncio.run(app.resolve_async(event, context))

examples/event_handler_rest/src/async_resolve_testing.py

@@ -0,0 +1,26 @@
+import asyncio
+import json
+
+
+def test_async_handler():
+    from async_resolve_getting_started import app  # (1)!
+
+    event = {
+        "httpMethod": "GET",
+        "path": "/todos/1",
+        "headers": {},
+        "queryStringParameters": None,
+        "pathParameters": {"todo_id": "1"},
+        "body": None,
+        "isBase64Encoded": False,
+        "requestContext": {"stage": "dev", "requestId": "test-id", "http": {"method": "GET", "path": "/todos/1"}},
+        "rawPath": "/todos/1",
+        "rawQueryString": "",
+        "routeKey": "GET /todos/{todo_id}",
+        "version": "2.0",
+    }
+
+    response = asyncio.run(app.resolve_async(event, {}))  # (2)!
+
+    assert response["statusCode"] == 200  # (3)!
+    assert json.loads(response["body"]) == {"todo_id": "1", "completed": False}