aws-lambda

Files (1)

• SKILL.md

SKILL.md (first lines)

# AWS Lambda

Conventions for AWS Lambda functions in Python.

## Related Patterns

For ingestion, ETL, and serialisation patterns that use Lambda handlers, also read:

- `../../catalogs/data-pipeline-patterns.md`

---

## General Rules

- Target the `python3.12` or `python3.13` runtime. Prefer `arm64` (Graviton) for
  lower price per GB-second unless a native dependency ships x86-only wheels.
- Keep handlers thin — business logic in separate, unit-testable modules.
- Set memory and timeout explicitly. API-backed handlers must finish within the API
  Gateway integration timeout (29 s default for REST APIs).
- Use Powertools for AWS Lambda (Python) v3 for logging, tracing, metrics, parsing,
  batch, idempotency, and parameters instead of hand-rolled equivalents.

---

## Handler Structure

Later snippets assume `import json`, `import os`, `import boto3`, and the
`LambdaContext` import below.

```python
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()    # service from POWERTOOLS_SERVICE_NAME; structured JSON
tracer = Tracer()    # needs X-Ray active tracing
metrics = Metrics()  # namespace from POWERTOOLS_METRICS_NAMESPACE


@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics
def handler(event: dict, context: LambdaContext) -> dict:
    # 1. Parse and validate input
    # 2. Call business logic
    # 3. Return structured response
    metrics.add_metric(name="RecordsProcessed", unit=MetricUnit.Count, value=1)
    ...
```

- Install the extras the code uses (`aws-lambda-powertools[tracer,parser]`) or
  attach the Powertools Lambda layer.
- Without Powertools, use stdlib `logging` with the function's log format set to
  JSON. Never `print`.

---

## Input Validation

- Validate every event at the boundary with a pydantic v2 model via Powertools
  `parser`; never index raw event keys deep in business logic.
- Return `400` for invalid API input; for queue/stream sources let the error fail
  the record (see Batch Processing).
- For multi-route APIs use `event_handler.APIGatewayRestResolver` /
  `APIGatewayHttpResolver` instead of hand-parsing the proxy event.

```python
from aws_lambda_powertools.utilities.parser import envelopes, parse
from pydantic import BaseModel, ValidationError


class OrderRequest(BaseModel):
    order_id: str
    quantity: int


def handler(event: dict, context: LambdaContext) -> dict:
    try:
        req = parse(event=event, model=OrderRequest, envelope=envelopes.ApiGatewayEnvelope)
    except ValidationError:
        req = None
    if req is None:  # the envelope returns None, not ValidationError, for a null body
        return {"statusCode": 400, "body": json.dumps({"error": "invalid request"})}
    ...
```

---

## Error Handling

- Catch specific exceptions; let unexpected ones propagate.
- Status codes matter only for API Gateway and function URLs. For async, SQS,
  Kinesis, DynamoDB Streams, and EventBridge sources, **raise** so Lambda retries
  and routes to the DLQ / on-failure destination — returning an error dict marks
  the event as processed.
- In API handlers, log with `logger.exception(...)` and return a generic `500`
  body; never return `str(e)` — it leaks internal details to the caller.

---

## Batch Processing

For SQS, Kinesis, and DynamoDB Streams, report per-record failures instead of
failing the whole batch: enable `ReportBatchItemFailures` on the event source
mapping and use Powertools `batch`.

```python
from aws_lambda_powertools.utilities.batch import (
    BatchProcessor, EventType, process_partial_response,
)
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

processor = BatchProcessor(event_type=EventType.SQS)


def record_handler(record: SQSRecord) -> None:
    ...  # raise to mark only this record as failed

From mabrosim/aet-content-pack, MIT.

More from mabrosim/aet-content-pack