0
# AWS Lambda Powertools Python
1

2
AWS Lambda Powertools Python is a comprehensive developer toolkit that implements serverless best practices for AWS Lambda functions. The package provides utilities for structured logging with Lambda context enrichment, distributed tracing, custom metrics, event handling, batch processing, parameter management, and many other serverless-specific features.
3

4
## Package Information
5

6
- **Package Name**: aws-lambda-powertools
7
- **Language**: Python 3.9+
8
- **Installation**: `pip install aws-lambda-powertools`
9

10
## Core Imports
11

12
```python
13
from aws_lambda_powertools import Logger, Metrics, Tracer
14
```
15

16
Specific utilities:
17

18
```python
19
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
20
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
21
from aws_lambda_powertools.utilities.parameters import get_parameter
22
from aws_lambda_powertools.utilities.batch import batch_processor
23
```
24

25
## Basic Usage
26

27
```python
28
from aws_lambda_powertools import Logger, Metrics, Tracer
29
from aws_lambda_powertools.utilities.typing import LambdaContext
30

31
# Initialize core observability utilities
32
logger = Logger()
33
metrics = Metrics()
34
tracer = Tracer()
35

36
@logger.inject_lambda_context(log_event=True)
37
@tracer.capture_lambda_handler
38
@metrics.log_metrics
39
def lambda_handler(event: dict, context: LambdaContext) -> dict:
40
    # Add custom metrics
41
    metrics.add_metric(name="ProcessedEvents", unit="Count", value=1)
42
    
43
    # Structure logging with context
44
    logger.info("Processing event", extra={"event_type": "example"})
45
    
46
    # Add tracing annotations
47
    tracer.put_annotation(key="operation", value="process_event")
48
    
49
    return {"statusCode": 200, "body": "Success"}
50
```
51

52
## Architecture
53

54
AWS Lambda Powertools follows a modular architecture with independent utilities:
55

56
- **Core Observability**: Logger, Metrics, and Tracer work together to provide structured logging, CloudWatch metrics, and X-Ray tracing
57
- **Event Handlers**: Framework-agnostic resolvers for API Gateway, ALB, AppSync, and other event sources
58
- **Utilities**: Standalone components for batch processing, parameter management, data parsing, and validation
59
- **Data Classes**: Type-safe representations of AWS event structures
60
- **Decorators**: Python decorators for seamless integration with existing Lambda functions
61

62
## Capabilities
63

64
### Core Observability
65

66
Essential observability utilities including structured logging with Lambda context enrichment, CloudWatch embedded metric format (EMF) for custom metrics, and AWS X-Ray integration for distributed tracing.
67

68
```python { .api }
69
class Logger:
70
    def __init__(
71
        self,
72
        service: str = None,
73
        level: str = "INFO",
74
        child: bool = False,
75
        sampling_rate: float = 0.0,
76
        stream: TextIO = None,
77
        logger_formatter: PowertoolsFormatter = None,
78
        logger_handler: logging.Handler = None,
79
        log_uncaught_exceptions: bool = False,
80
        json_serializer: Callable[[Dict], str] = None,
81
        json_deserializer: Callable[[Union[Dict, str, bool, int, float]], str] = None,
82
        json_default: Callable[[Any], Any] = None,
83
        datefmt: str = None,
84
        use_datetime_directive: bool = False,
85
        log_record_order: List[str] = None,
86
        utc: bool = False,
87
        use_rfc3339: bool = False,
88
    ): ...
89

90
class Metrics:
91
    def __init__(
92
        self,
93
        service: str = None,
94
        namespace: str = None,
95
        metadata: Dict[str, Any] = None,
96
        default_dimensions: Dict[str, str] = None,
97
    ): ...
98

99
class Tracer:
100
    def __init__(
101
        self,
102
        service: str = None,
103
        disabled: bool = False,
104
        auto_patch: bool = True,
105
        patch_modules: List[str] = None,
106
        provider: BaseProvider = None,
107
    ): ...
108
```
109

110
[Core Observability](./core-observability.md)
111

112
### Event Handlers
113

114
Framework for building Lambda functions that handle HTTP events from API Gateway, Application Load Balancer, AppSync GraphQL, and other AWS services with automatic request/response serialization.
115

116
```python { .api }
117
class APIGatewayRestResolver:
118
    def __init__(
119
        self,
120
        cors: CORSConfig = None,
121
        debug: bool = None,
122
        serializer: Callable[[Dict], str] = None,
123
        strip_prefixes: List[str] = None,
124
        enable_validation: bool = False,
125
    ): ...
126
    
127
    def get(self, rule: str, **kwargs): ...
128
    def post(self, rule: str, **kwargs): ...
129
    def put(self, rule: str, **kwargs): ...
130
    def delete(self, rule: str, **kwargs): ...
131

132
class ALBResolver:
133
    def __init__(
134
        self,
135
        cors: CORSConfig = None,
136
        debug: bool = None,
137
        serializer: Callable[[Dict], str] = None,
138
        strip_prefixes: List[str] = None,
139
        enable_validation: bool = False,
140
    ): ...
141

142
class AppSyncResolver:
143
    def __init__(self, debug: bool = None): ...
144
    def resolver(self, type_name: str = "*", field_name: str = None): ...
145
```
146

147
[Event Handlers](./event-handlers.md)
148

149
### Batch Processing
150

151
Utilities for processing AWS SQS, DynamoDB Streams, and Kinesis records with built-in error handling, partial failure support, and automatic retries.
152

153
```python { .api }
154
def batch_processor(
155
    record_handler: Callable[[Dict], Any],
156
    processor: BatchProcessor,
157
    context: LambdaContext = None,
158
): ...
159

160
def async_batch_processor(
161
    record_handler: Callable[[Dict], Awaitable[Any]],
162
    processor: AsyncBatchProcessor,
163
    context: LambdaContext = None,
164
): ...
165

166
class BatchProcessor:
167
    def __init__(
168
        self,
169
        event_type: EventType,
170
        model: BaseModel = None,
171
        batch_length_quota_mb: int = 6,
172
    ): ...
173
```
174

175
[Batch Processing](./batch-processing.md)
176

177
### Data Classes
178

179
Type-safe data classes representing AWS event structures for Lambda triggers including API Gateway, S3, DynamoDB, Kinesis, SQS, SNS, and CloudWatch events.
180

181
```python { .api }
182
class APIGatewayProxyEvent:
183
    @property
184
    def body(self) -> str | None: ...
185
    @property
186
    def json_body(self) -> Any: ...
187
    @property
188
    def headers(self) -> Dict[str, str]: ...
189
    @property
190
    def query_string_parameters(self) -> Dict[str, str] | None: ...
191

192
class SQSEvent:
193
    @property
194
    def records(self) -> List[SQSRecord]: ...
195

196
class DynamoDBStreamEvent:
197
    @property
198
    def records(self) -> List[DynamoDBRecord]: ...
199
```
200

201
[Data Classes](./data-classes.md)
202

203
### Parameters
204

205
Retrieve and cache parameters from AWS Systems Manager Parameter Store, AWS Secrets Manager, and AWS AppConfig with automatic caching and transformation support.
206

207
```python { .api }
208
def get_parameter(
209
    name: str,
210
    decrypt: bool = True,
211
    max_age: int = 5,
212
    transform: str = None,
213
    force_fetch: bool = False,
214
    **sdk_options,
215
) -> str: ...
216

217
def get_secret(
218
    name: str,
219
    version_id: str = None,
220
    version_stage: str = None,
221
    max_age: int = 5,
222
    transform: str = None,
223
    force_fetch: bool = False,
224
    **sdk_options,
225
) -> str: ...
226

227
def get_app_config(
228
    name: str,
229
    environment: str,
230
    application: str,
231
    max_age: int = 5,
232
    transform: str = None,
233
    force_fetch: bool = False,
234
    **sdk_options,
235
) -> bytes | str: ...
236
```
237

238
[Parameters](./parameters.md)
239

240
### Parser
241

242
Event parsing and validation using Pydantic models with built-in envelopes for extracting business logic from AWS event sources.
243

244
```python { .api }
245
def event_parser(
246
    model: BaseModel,
247
    envelope: BaseEnvelope = None,
248
) -> Callable: ...
249

250
def parse(
251
    event: Dict[str, Any],
252
    model: BaseModel,
253
    envelope: BaseEnvelope = None,
254
) -> Any: ...
255

256
class BaseEnvelope:
257
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any: ...
258
```
259

260
[Parser](./parser.md)
261

262
### Feature Flags & Idempotency
263

264
Feature flags with rule engine support from AWS AppConfig and idempotency patterns to prevent duplicate processing of events.
265

266
```python { .api }
267
class FeatureFlags:
268
    def __init__(
269
        self,
270
        store: StoreProvider,
271
        logger: Logger = None,
272
    ): ...
273
    
274
    def evaluate(
275
        self,
276
        name: str,
277
        context: Dict[str, Any] = None,
278
        default: Any = False,
279
    ) -> bool | Any: ...
280

281
def idempotent(
282
    persistence_store: BasePersistenceLayer,
283
    config: IdempotencyConfig = None,
284
) -> Callable: ...
285

286
def idempotent_function(
287
    data_keyword_argument: str,
288
    persistence_store: BasePersistenceLayer,
289
    config: IdempotencyConfig = None,
290
) -> Callable: ...
291
```
292

293
[Feature Flags & Idempotency](./feature-flags.md)
294

295
### Utilities
296

297
Additional utilities including data masking, streaming, serialization, validation, JMESPath operations, and Kafka consumer helpers.
298

299
```python { .api }
300
class DataMasking:
301
    def __init__(self, provider: BaseProvider): ...
302
    def erase(self, data: Any, fields: List[str] = None) -> Any: ...
303

304
class S3Object:
305
    def __init__(self, bucket: str, key: str, **kwargs): ...
306
    def transform(self, transform: BaseTransform) -> "S3Object": ...
307

308
def validate(
309
    event: Dict[str, Any],
310
    schema: Dict[str, Any],
311
    envelope: str = None,
312
) -> Dict[str, Any]: ...
313
```
314

315
[Utilities](./utilities.md)
316

317
## Core Types
318

319
```python { .api }
320
from typing import Dict, Any, List, Optional, Union, Callable, Awaitable
321
from aws_lambda_powertools.utilities.typing import LambdaContext
322

323
class LambdaContext:
324
    """AWS Lambda context object"""
325
    function_name: str
326
    function_version: str
327
    invoked_function_arn: str
328
    memory_limit_in_mb: int
329
    remaining_time_in_millis: int
330
    request_id: str
331
    log_group_name: str
332
    log_stream_name: str
333
    identity: Any
334
    client_context: Any
335

336
# Metric and logging types
337
MetricUnit = Literal[
338
    "Seconds", "Microseconds", "Milliseconds", "Bytes", "Kilobytes", 
339
    "Megabytes", "Gigabytes", "Terabytes", "Bits", "Kilobits", 
340
    "Megabits", "Gigabits", "Terabits", "Percent", "Count", 
341
    "Bytes/Second", "Kilobytes/Second", "Megabytes/Second", 
342
    "Gigabytes/Second", "Terabytes/Second", "Bits/Second", 
343
    "Kilobits/Second", "Megabits/Second", "Gigabits/Second", 
344
    "Terabits/Second", "Count/Second", "None"
345
]
346

347
MetricResolution = Literal[1, 60]
348

349
# Event handler types
350
class CORSConfig:
351
    def __init__(
352
        self,
353
        allow_origin: str = "*",
354
        allow_headers: List[str] = None,
355
        allow_methods: List[str] = None,
356
        expose_headers: List[str] = None,
357
        max_age: int = None,
358
        allow_credentials: bool = False,
359
    ): ...
360

361
class Response:
362
    def __init__(
363
        self,
364
        status_code: int,
365
        content_type: str = None,
366
        body: str = None,
367
        headers: Dict[str, str] = None,
368
        cookies: List[str] = None,
369
    ): ...
370

371
# Batch processing types
372
EventType = Literal["SQS", "KinesisDataStreams", "DynamoDBStreams"]
373

374
class SuccessResponse:
375
    def __init__(self, **kwargs): ...
376

377
class FailureResponse:
378
    def __init__(self, **kwargs): ...
379

380
class ExceptionInfo:
381
    def __init__(self, exception: Exception, record: Dict[str, Any]): ...
382

383
# Parameter and configuration types
384
class GetParameterError(Exception): ...
385
class TransformParameterError(Exception): ...
386
class ConfigurationStoreError(Exception): ...
387

388
# Validation types  
389
class ValidationError(Exception): ...
390
class SchemaValidationError(Exception): ...
391
class InvalidSchemaFormatError(Exception): ...
392
class InvalidEnvelopeExpressionError(Exception): ...
393

394
# Parser types
395
class BaseModel:
396
    """Pydantic BaseModel re-export for parser functionality"""
397
    pass
398

399
def Field(**kwargs) -> Any:
400
    """Pydantic Field function re-export"""
401
    pass
402

403
# Idempotency types
404
class IdempotencyConfig:
405
    def __init__(
406
        self,
407
        event_key_jmespath: str = None,
408
        payload_validation_jmespath: str = None,
409
        raise_on_no_idempotency_key: bool = False,
410
        expires_after_seconds: int = 3600,
411
        use_local_cache: bool = False,
412
        local_cache_max_items: int = 256,
413
        hash_function: str = "md5",
414
        lambda_context: LambdaContext = None,
415
    ): ...
416

417
# Feature flags types
418
RuleAction = Literal["ALLOW", "DENY"]
419

420
class SchemaValidator:
421
    def __init__(self, schema: Dict[str, Any]): ...
422
    def validate(self, data: Dict[str, Any]) -> Dict[str, Any]: ...
423
```