| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
We're really happy to say the Circuit Breaker utility is now GA. It shipped as alpha in 3.31.0 to collect real-world feedback, and that's exactly what happened, so it's time to drop the _alpha suffix and call it stable.
A big thank you to everyone who used it, reported issues and sent fixes during the alpha. It's in much better shape because of you.
The only change you need to make is the import path. circuit_breaker_alpha becomes circuit_breaker:
from aws_lambda_powertools.utilities.circuit_breaker import circuit_breaker
from aws_lambda_powertools.utilities.circuit_breaker.persistence import (
CircuitBreakerDynamoDBPersistence,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
persistence = CircuitBreakerDynamoDBPersistence(table_name="CircuitBreakerState")
@circuit_breaker(name="payment-backend", persistence_store=persistence)
def charge(order: dict) -> dict:
return payment_api.charge(order)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return charge(event)Everything else stays the same. Same API, same defaults (open after 5 consecutive failures, probe after 30 seconds, close after 3 probe successes), same behaviour. If you were using it during the alpha, update the import and you're done.
You'll notice we jumped from 3.31.1 straight to 3.34.0. Nothing to worry about: our release pipeline accidentally skipped 3.32.0 and 3.33.0, so they were never built or published. No package went to PyPI and no Lambda layer was published for either of them, so there's no impact for anyone. Everything that was meant to go out in those two releases is included here in 3.34.0.
@DebadityaHait, @leandrodamascena, @vishwakt and @dependabot[bot]
This release fixes bugs across Circuit Breaker, Event Handler, Parser, and the Event Source data classes. Super thanks to @Iamrodos for running the Circuit Breaker alpha for real and finding the concurrency and configuration issues fixed here.
@Iamrodos, @amin-farjadi, @dependabot[bot], @exg, @github-actions[bot], @leandrodamascena, @stenczelt, dependabot[bot] and github-actions[bot]
This release adds a new Circuit Breaker utility (in alpha) that stops your Lambda from sending requests to an unhealthy downstream and gives it time to recover. We also made parameter validation in the Event Handler more flexible, so any Pydantic Field annotation now works with any parameter type.
A huge thanks to everyone who helped shape the Circuit Breaker RFC and reviewed this release!
When a downstream service is failing, retries and Lambda's scaling only make it worse: more clients sending requests to something that is already down. The Circuit Breaker stops sending traffic to an unhealthy dependency, then probes it to see when it is safe to resume.
It ships as circuit_breaker_alpha on purpose. We want about a month of real-world feedback before we lock the public API and promote it to GA.
The smallest setup is a persistence store and a name. You wrap the function that makes the downstream call:
from aws_lambda_powertools.utilities.circuit_breaker_alpha import circuit_breaker
from aws_lambda_powertools.utilities.circuit_breaker_alpha.persistence import (
CircuitBreakerDynamoDBPersistence,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
persistence = CircuitBreakerDynamoDBPersistence(table_name="CircuitBreakerState")
@circuit_breaker(name="payment-backend", persistence_store=persistence)
def charge(order: dict) -> dict:
return payment_api.charge(order)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return charge(event)With no config, sensible defaults apply: open after 5 failures in a row, probe after 30s, close after 3 successes, and treat any exception as a failure. A few things make it a good fit for Lambda:
When the circuit is open, you decide what happens to the rejected request with an on_circuit_open callback (buffer it, drop it, return a cached value), or let it raise CircuitBreakerOpenError. You can also watch state changes with an on_transition hook to emit your own metrics.
import json
from uuid import uuid4
from aws_lambda_powertools.utilities.circuit_breaker_alpha import circuit_breaker
from aws_lambda_powertools.utilities.circuit_breaker_alpha.persistence import (
CircuitBreakerDynamoDBPersistence,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
persistence = CircuitBreakerDynamoDBPersistence(table_name="CircuitBreakerState")
def buffer_payload(payload: dict, circuit) -> None:
# Circuit is OPEN. The call never ran, so the payload is yours to handle.
s3.put_object(Bucket="payment-overflow", Key=f"{circuit.name}/{uuid4()}", Body=json.dumps(payload))
@circuit_breaker(
name="payment-backend",
persistence_store=persistence,
on_circuit_open=buffer_payload,
)
def charge(order: dict) -> dict:
return payment_api.charge(order)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return charge(event)You can now use any Pydantic Field annotation with any parameter location (Path, Query, Header, Body), the same way it works inside a model. Before, only Field(discriminator=...) with Body() was supported.
from typing import Annotated
from pydantic import Field
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.openapi.params import Query
from aws_lambda_powertools.utilities.typing import LambdaContext
app = APIGatewayHttpResolver(enable_validation=True)
@app.get("/count")
def get_count(n: Annotated[int, Field(gt=0), Query]): # gt=0 enforced, 422 on n <= 0
return {"count": n}
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)Last but not least, thanks to everyone who reported issues and helped us improve this release.
@dependabot[bot], @github-actions[bot], @leandrodamascena, dependabot[bot] and github-actions[bot]
This release adds a custom serializer option to the BedrockAgentResolver, plus documentation improvements and bug fixes across the Event Handler.
A huge thanks to @kimnamu, @Avinm and @hirenkumar-n-dholariya for their contributions!
You can now pass your own serializer to BedrockAgentResolver, the same way the other resolvers already allow. This lets you control exactly how responses are serialized to JSON, for example to handle custom types or tune the output format.
import json
from decimal import Decimal
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.utilities.typing import LambdaContext
def custom_serializer(obj: dict) -> str:
return json.dumps(obj, default=str)
app = BedrockAgentResolver(serializer=custom_serializer)
@app.get("/price", description="Returns the current price")
def get_price() -> dict:
return {"price": Decimal("9.99")}
def lambda_handler(event: dict, context: LambdaContext):
return app.resolve(event, context)@Avinm, @Sujit-1509, @dependabot[bot], @derdelean, @github-actions[bot], @hirenkumar-n-dholariya, @kimnamu, @leandrodamascena, dependabot[bot] and github-actions[bot]
We're thrilled to announce native async resolution for the Event Handler. Write async def route handlers, call await app.resolve_async(event, context), and mix sync/async middlewares. Everything runs natively on the event loop.
We also fixed OpenAPI schema generation for Pydantic @computed_field, a deadlock when sync middlewares raised before calling next(), and ALB returning 422 when response body is None.
A huge thanks to @hirenkumar-n-dholariya, @amrabed, and @catarinacps for their contributions!
You can now define async route handlers and resolve them without blocking the event loop.
Even though the Lambda handler itself is sync, you can use asyncio.run(app.resolve_async(event, context)) to fan out multiple I/O calls concurrently with asyncio.gather or call async libraries directly.
import asyncio
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.utilities.typing import LambdaContext
app = APIGatewayHttpResolver()
async def get_orders(user_id: str) -> list:
...
async def get_profile(user_id: str) -> dict:
...
@app.get("/dashboard/<user_id>")
async def get_dashboard(user_id: str):
orders, profile = await asyncio.gather(
get_orders(user_id),
get_profile(user_id),
)
return {"orders": orders, "profile": profile}
def lambda_handler(event: dict, context: LambdaContext):
return asyncio.run(app.resolve_async(event, context))Sync and async middlewares work together seamlessly. Sync middlewares are bridged to the event loop in a background thread, so you don't need to rewrite existing middleware to adopt async handlers.
Pydantic @computed_field properties now appear in generated OpenAPI schemas. Previously they were excluded because we always used mode="validation" when generating JSON schemas.
from pydantic import BaseModel, computed_field
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
app = APIGatewayHttpResolver(enable_validation=True)
class Order(BaseModel):
price: float
quantity: int
@computed_field
@property
def total(self) -> float:
return self.price * self.quantity
@app.get("/order")
def get_order() -> Order:
return Order(price=10.0, quantity=3)Last but not least, thanks to @chriselion and @avplab for reporting bugs in our Event Handler resolvers.
@amrabed, @catarinacps, @dependabot[bot], @github-actions[bot], @hirenkumar-n-dholariya, @leandrodamascena, dependabot[bot] and github-actions[bot]
This release brings dependency injection, an enriched Request object, OpenAPI improvements, and internal refactoring to the Event Handler.
A huge thanks to @JustinBerger, @Iamrodos, and @ran-isenberg for their contributions!
You can now use Depends() to declare typed dependencies directly in route handler signatures: no decorators, no global state. Dependencies are resolved automatically, cached per invocation, and support nested dependency trees.
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.depends 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"]For testing, swap any dependency without monkeypatching:
app.dependency_overrides[get_dynamodb_table] = lambda: mock_tableThe Request object now exposes resolved_event and context, enabling dependency functions to access the full Powertools event and bridge data with middleware.
from typing_extensions import Annotated
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver, Response
from aws_lambda_powertools.event_handler.depends import Depends
from aws_lambda_powertools.event_handler.request import Request
app = APIGatewayHttpResolver()
# Middleware handles auth: can return HTTP responses (redirects, 401s)
def auth_middleware(app, next_middleware):
token = app.current_event.headers.get("authorization", "")
if not token:
return Response(status_code=401, body="Unauthorized")
app.append_context(user={"id": "user-123", "role": "admin"})
return next_middleware(app)
app.use(middlewares=[auth_middleware])
# Depends() reads what middleware wrote via request.context
def get_current_user(request: Request) -> dict:
return request.context["user"]
@app.get("/admin/dashboard")
def admin_dashboard(user: Annotated[dict, Depends(get_current_user)]):
return {"message": f"Welcome {user['id']}", "role": user["role"]}You can now set the default response status code directly on route decorators. This is reflected in the generated OpenAPI schema and Swagger UI.
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
app = APIGatewayRestResolver(enable_validation=True)
@app.post(
"/todos",
summary="Creates a new todo item",
status_code=201,
tags=["Todos"],
)
def create_todo(title: str) -> dict:
return {"id": 1, "title": title}@JustinBerger, @dependabot[bot], @github-actions[bot], @leandrodamascena, dependabot[bot] and github-actions[bot]
In this release, we focused on the Event Handler utility - we added three new features and shipped several important bug fixes across Event Handler and Idempotency.
A huge thanks to @oyiz-michael, @siwyd, @abhu85, and @danjhd for their contributions!
You can now handle file uploads in your API endpoints with full OpenAPI validation and Swagger UI support. If using Swagger, it renders a file picker automatically. A special thanks to @oyiz-michael for starting the initial work on this feature.
from typing import Annotated
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.openapi.params import File, Form, UploadFile
app = APIGatewayRestResolver(enable_validation=True)
app.enable_swagger(path="/swagger")
@app.post("/upload")
def upload(
file_data: Annotated[UploadFile, File(description="CSV file")],
separator: Annotated[str, Form(description="CSV separator")] = ",",
):
return {
"filename": file_data.filename,
"content_type": file_data.content_type,
"file_size": len(file_data),
}You can receive files as raw bytes (Annotated[bytes, File()]) or as an UploadFile object with filename and content type metadata.
You can now use cookies as typed, validated parameters in your API endpoints - just like Query(), Header(), or Form(). The OpenAPI schema generates in: cookie parameters automatically, and validation works across all resolver types.
from typing import Annotated
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.openapi.params import Cookie
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/me")
def get_me(
session_id: Annotated[str, Cookie(description="Session identifier")],
theme: Annotated[str, Cookie(description="UI theme")] = "light",
):
return {"session_id": session_id, "theme": theme}We added a Request object that gives middleware and route handlers access to the resolved route pattern, Powertools-extracted path parameters, HTTP method, headers, query parameters, and body. Previously, middleware only had access to app.current_event, which returns raw API Gateway parameters (e.g. {"proxy": "users/123"} for {proxy+} routes) instead of the resolved ones.
You can access the Request object in two ways:
In middleware via app.request:
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
app = APIGatewayRestResolver()
def auth_middleware(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
req = app.request
route = req.route # "/users/{user_id}"
path_params = req.path_parameters # {"user_id": "123"}
method = req.method # "GET"
# auth logic here...
return next_middleware(app)
app.use(middlewares=[auth_middleware])In route handlers via type annotation:
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Request
app = APIGatewayRestResolver()
@app.get("/users/<user_id>")
def get_user(user_id: str, request: Request):
user_agent = request.headers.get("user-agent")
return {"id": user_id, "route": request.route, "user_agent": user_agent}@abhu85, @danjhd, @dependabot[bot], @github-actions[bot], @leandrodamascena, @oyiz-michael, @ran-isenberg, @siwyd, dependabot[bot] and github-actions[bot]
In this release, we are pleased to announce a new utility for interacting with the Lambda Metadata Service, allowing you to easily retrieve information about the Lambda function, such as the Availability Zone ID.
A huge thanks to @acascell, @shaked-lokits, @maxrabin, and @amin-farjadi, for their contributions 🚀🌟
You can now use get_lambda_metadata() function fetch metadata from the AWS Lambda Metadata endpoint, such as the Availability Zone ID. Results are cached for the sandbox lifetime and the utility automatically returns empty metadata outside of Lambda, so your code works seamlessly in local development and testing.
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.metadata import LambdaMetadata, get_lambda_metadata
from aws_lambda_powertools.utilities.typing import LambdaContext
logger = Logger()
def lambda_handler(event: dict, context: LambdaContext) -> dict:
metadata: LambdaMetadata = get_lambda_metadata()
az_id = metadata.availability_zone_id # e.g., "use1-az1"
logger.append_keys(az_id=az_id)
logger.info("Processing request")
return {"az_id": az_id}@acascell, @amin-farjadi, @dependabot[bot], @github-actions[bot], @leandrodamascena, @shaked-lokits, dependabot[bot] and github-actions[bot]
This release introduces per-route validation support in event handler, durable context support for logger and metric decorators, multiple dimension sets in metrics, S3 IntelligentTiering event support, and a URL-decode flag for ALB query parameters. We also shipped several important bug fixes across logger, parameters, event handler, and typing.
A huge thanks to @oyiz-michael, @chriselion, @maxrabin, @facu-01, and @Iamrodos, for their contributions 🚀🌟
You can now enable or disable validation on individual routes. This is useful when migrating incrementally - enable validation globally and opt-out specific legacy routes, or vice versa.
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from pydantic import BaseModel
app = APIGatewayRestResolver(enable_validation=True)
class Todo(BaseModel):
title: str
completed: bool
# This route has validation enabled (inherits from resolver)
@app.post("/todos")
def create_todo(todo: Todo) -> dict:
return {"title": todo.title}
# This route opts out of validation
@app.get("/legacy", enable_validation=False)
def legacy_endpoint():
return {"message": "no validation here"}Logger and Metrics decorators now handle AWS Lambda Durable Context automatically. When a durable function replays, the decorators unwrap the DurableContext to access the underlying Lambda context - no changes needed in your code.
from aws_lambda_powertools import Logger, Metrics
from aws_lambda_powertools.metrics import MetricUnit
logger = Logger()
metrics = Metrics()
@logger.inject_lambda_context # automatically handles DurableContext
@metrics.log_metrics # automatically handles DurableContext
def lambda_handler(event, context):
logger.info("Processing event")
metrics.add_metric(name="InvocationCount", unit=MetricUnit.Count, value=1)
return {"statusCode": 200}You can now publish metrics with multiple dimension sets using add_dimensions(). Each call creates a new dimension set in the CloudWatch EMF output.
from aws_lambda_powertools import Metrics
from aws_lambda_powertools.metrics import MetricUnit
metrics = Metrics()
@metrics.log_metrics
def lambda_handler(event, context):
metrics.add_metric(name="OrderCount", unit=MetricUnit.Count, value=1)
# Each call creates a separate dimension set
metrics.add_dimensions(environment="prod", region="us-east-1")
metrics.add_dimensions(service="orders", team="backend")A new decode_query_parameters flag in ALBResolver automatically URL-decodes query parameter keys and values.
from aws_lambda_powertools.event_handler import ALBResolver
app = ALBResolver(decode_query_parameters=True)
@app.get("/search")
def search():
# Query params are automatically URL-decoded
query = app.current_event.query_string_parameters
return {"query": query}This release adds support for Lambda durable function replay in idempotency, a new parser model for DynamoDB Stream on-failure destinations, and a fix for batch processor. We've also dropped Python 3.9 support.
A super thanks to @ConnorKirk for implementing the idempotency replay feature and @exg for the batch processor fix and DynamoDB Stream parser model 🚀🌟
AWS Lambda durable functions enable you to build resilient multi-step applications that can execute for up to one year while maintaining reliable progress despite interruptions. When a durable function resumes from a wait point or interruption, the system performs replay - running your code from the beginning but skipping completed checkpoints.
This release adds support for durable function replay in idempotency. When using the @idempotent decorator on lambda_handler, replay is automatically detected from the DurableContext - no manual configuration needed.
from aws_lambda_powertools.utilities.idempotency import idempotent, DynamoDBPersistenceLayer, IdempotencyConfig
persistence_layer = DynamoDBPersistenceLayer(table_name="idempotency_store")
config = IdempotencyConfig(event_key_jmespath="body")
@idempotent(config=config, persistence_store=persistence_layer)
def lambda_handler(event, context):
# Replay is automatically detected when context is a DurableContext
# During replay, INPROGRESS records are handled gracefully
return {"statusCode": 200}A new parser model DynamoDBStreamLambdaOnFailureDestinationModel is now available for parsing DynamoDB Stream on-failure destination events.
from aws_lambda_powertools.utilities.parser import parse
from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamLambdaOnFailureDestinationModel
def lambda_handler(event, context):
parsed = parse(event=event, model=DynamoDBStreamLambdaOnFailureDestinationModel)
batch_info = parsed.ddb_stream_batch_info
print(f"Failed batch from shard: {batch_info.shard_id}")
print(f"Stream ARN: {batch_info.stream_arn}")| Back | FazBrowse Home | New Git URL |