-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
148 lines (123 loc) · 5.42 KB
/
app.py
File metadata and controls
148 lines (123 loc) · 5.42 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""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.
"""
import os
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.api_gateway import CORSConfig
from aws_lambda_powertools.event_handler.exceptions import InternalServerError
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags
from aws_lambda_powertools.utilities.idempotency import (
DynamoDBPersistenceLayer,
idempotent,
)
from aws_lambda_powertools.utilities.idempotency.config import IdempotencyConfig
from aws_lambda_powertools.utilities.parameters import get_parameter
from aws_lambda_powertools.utilities.typing import LambdaContext
from pydantic import BaseModel
logger = Logger()
tracer = Tracer()
metrics = Metrics()
# enable_validation=True wires Pydantic into the resolver. Request bodies and
# response return types are validated against their model annotations, and
# those same models drive the OpenAPI schema read by scripts/generate_openapi.py.
# We deliberately do NOT call app.enable_swagger() here — exposing the spec at
# runtime would publish the full API surface to any caller. The spec is
# instead rendered into Zensical at documentation-build time.
app = APIGatewayRestResolver(
cors=CORSConfig(allow_origin="*", max_age=300),
enable_validation=True,
)
# Idempotency setup
persistence_layer = DynamoDBPersistenceLayer(
table_name=os.environ.get("IDEMPOTENCY_TABLE_NAME", ""),
)
idempotency_config = IdempotencyConfig(
event_key_jmespath="requestContext.requestId",
expires_after_seconds=3600,
)
# Feature Flags setup
app_config_store = AppConfigStore(
environment=os.environ.get("APPCONFIG_ENV_NAME", ""),
application=os.environ.get("APPCONFIG_APP_NAME", ""),
name=os.environ.get("APPCONFIG_PROFILE_NAME", ""),
)
feature_flags = FeatureFlags(store=app_config_store)
class HelloResponse(BaseModel):
"""Response body for GET /hello."""
message: str
@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 — required, raise 500 on failure
param_name = os.environ.get("GREETING_PARAM_NAME", "/HelloWorld/greeting")
try:
greeting = get_parameter(param_name)
except Exception as exc:
logger.exception("Failed to fetch greeting from SSM", param_name=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
try:
enhanced = feature_flags.evaluate(name="enhanced_greeting", default=False)
except Exception:
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)
@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
@idempotent(config=idempotency_config, persistence_store=persistence_layer)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
"""Lambda entry point.
Resolves the API Gateway event through the router and returns the result.
Decorated with Powertools Logger, Tracer, Metrics, and Idempotency.
Args:
event: API Gateway Lambda proxy event.
context: Lambda runtime context.
Returns:
dict: API Gateway Lambda proxy response.
"""
return app.resolve(event, context) # type: ignore[no-any-return]