Hello World Lambda function using AWS Lambda Powertools.
This module implements a serverless API endpoint that returns a greeting message.
It demonstrates the use of Powertools utilities including structured logging,
X-Ray tracing, CloudWatch metrics, idempotency, SSM parameters, feature flags,
Pydantic-backed request/response validation (with an OpenAPI spec generated
at documentation-build time — see scripts/generate_openapi.py), and Event Source
Data Classes.
HelloResponse
Bases: BaseModel
Response body for GET /hello.
hello()
Handle GET /hello requests.
Fetches the greeting from SSM Parameter Store, checks the enhanced_greeting
feature flag, emits a CloudWatch metric, and logs request metadata from
the API Gateway event.
Returns:
| Name | Type |
Description |
HelloResponse |
HelloResponse
|
Validated response model with a message field.
|
Source code in lambda/app.py
| @app.get(
"/hello",
summary="Return a greeting",
description=(
"Returns the greeting string configured in SSM Parameter Store. "
"When the `enhanced_greeting` AppConfig feature flag is enabled for "
"the caller's source IP, the response includes the feature flag's "
"configured suffix."
),
response_description="A JSON object containing the resolved greeting.",
tags=["Greeting"],
)
@tracer.capture_method
def hello() -> HelloResponse:
"""Handle GET /hello requests.
Fetches the greeting from SSM Parameter Store, checks the enhanced_greeting
feature flag, emits a CloudWatch metric, and logs request metadata from
the API Gateway event.
Returns:
HelloResponse: Validated response model with a ``message`` field.
"""
metrics.add_metric(name="HelloRequests", unit=MetricUnit.Count, value=1)
# Access typed event data via Event Source Data Classes
event: APIGatewayProxyEvent = app.current_event
source_ip = event.request_context.identity.source_ip
user_agent = event.request_context.identity.user_agent
request_id = event.request_context.request_id
logger.info(
"Request received",
source_ip=source_ip,
user_agent=user_agent,
request_id=request_id,
)
# Fetch greeting from SSM Parameter Store. Powertools wraps boto3 errors
# (ClientError, BotoCoreError) as GetParameterError; catch only that so
# truly unexpected exceptions propagate to Powertools' default handler
# and surface with the right type in metrics and X-Ray.
# max_age=300 raises Powertools' in-memory TTL from its 5-second default
# so warm containers reuse the value for 5 minutes between SSM calls.
# The greeting changes via deployment, not at runtime, so a longer TTL
# is safe and meaningfully reduces SSM API spend at higher RPS.
try:
greeting = get_parameter(GREETING_PARAM_NAME, max_age=300)
except GetParameterError as exc:
logger.exception("Failed to fetch greeting from SSM", param_name=GREETING_PARAM_NAME)
raise InternalServerError("Failed to fetch greeting") from exc
logger.info("Greeting fetched from parameter store", greeting=greeting)
# Check feature flag — non-critical, fall back to default on failure.
# Pass source_ip + user_agent as context so AppConfig rules can match on
# them (the route's docstring promises IP-based gating; without context
# the rule engine can never see the values to evaluate against).
# Catch only the Powertools FeatureFlags exception types — programming
# errors (TypeError, AttributeError) intentionally propagate so they
# surface as bugs in metrics rather than being silently absorbed by the
# fallback path.
try:
enhanced = feature_flags.evaluate(
name="enhanced_greeting",
context={"source_ip": source_ip, "user_agent": user_agent},
default=False,
)
except (ConfigurationStoreError, SchemaValidationError, StoreClientError):
logger.warning("Feature flag evaluation failed, falling back to default")
enhanced = False
if enhanced:
message = f"{greeting} - enhanced mode enabled"
logger.info("Enhanced greeting enabled")
else:
message = greeting
return HelloResponse(message=message)
|
lambda_handler(event, context)
Lambda entry point.
Resolves the API Gateway event through the router and returns the result.
Decorated with Powertools Logger, Tracer, Metrics; the inner function
handles Idempotency so a missing Idempotency-Key header surfaces as a 400
instead of an unhandled 500.
Parameters:
| Name |
Type |
Description |
Default |
event
|
dict[str, Any]
|
API Gateway Lambda proxy event.
|
required
|
context
|
LambdaContext
|
|
required
|
Returns:
| Name | Type |
Description |
dict |
dict[str, Any]
|
API Gateway Lambda proxy response.
|
Source code in lambda/app.py
| @logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
"""Lambda entry point.
Resolves the API Gateway event through the router and returns the result.
Decorated with Powertools Logger, Tracer, Metrics; the inner function
handles Idempotency so a missing Idempotency-Key header surfaces as a 400
instead of an unhandled 500.
Args:
event: API Gateway Lambda proxy event.
context: Lambda runtime context.
Returns:
dict: API Gateway Lambda proxy response.
"""
# cast() restores the return type after @idempotent erases it. Powertools'
# app.resolve() is well-typed in .venv-lambda, but the @idempotent wrapper
# passes return values through as Any; .venv (CDK side, no Powertools)
# already sees the function as Any. The cast is a no-op at runtime.
try:
return cast(dict, _resolve_with_idempotency(event, context))
except IdempotencyKeyError:
logger.warning("Request rejected: missing Idempotency-Key header")
return {
"statusCode": 400,
"headers": {"Content-Type": "application/json"},
"body": '{"message":"Idempotency-Key header is required"}',
"isBase64Encoded": False,
}
|