Demo of idempotent (More complete examples)

app.py

import json
import os
import time
from typing import Callable

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
from aws_lambda_powertools.utilities.idempotency import (
    IdempotencyConfig,
    DynamoDBPersistenceLayer,
    idempotent,
)
from aws_lambda_powertools.utilities.idempotency.exceptions import (
    IdempotencyAlreadyInProgressError,
    IdempotencyKeyError,
    IdempotencyPersistenceLayerError,
)
from aws_lambda_powertools.utilities.typing import LambdaContext


tracer = Tracer()
logger = Logger()
metrics = Metrics()
config = IdempotencyConfig(
    event_key_jmespath="powertools_json(body).order_key",  # Use the body.order_key as the key
    raise_on_no_idempotency_key=True,  # Raise an error if order_key was not found
    use_local_cache=True,  # Local caching
    local_cache_max_items=512,  # Cache up to 512 responses in local in memory cache
    expires_after_seconds=24 * 60 * 60,  # Expire idempotent response for up to 24 hours
)
persistence_layer = DynamoDBPersistenceLayer(
    table_name=os.environ["IDEMPOTENCY_TABLE_NAME"],
)


class PaymentServiceError(Exception):
    """Some generic payment error"""


@tracer.capture_method
def create_payment(payment: dict) -> dict:
    """
    Create payment call

    Parameters
    ----------
    payment :
        Payment request
    Returns
    -------
    dict
        Returns payment response
    """
    order_key = payment["order_key"]

    # Simulate failure
    if order_key == "FAILURE":
        raise PaymentServiceError("Failed to make payment")

    # Simulate a slow transaction
    time.sleep(10)

    metrics.add_metric(name="PaymentSuccessful", unit=MetricUnit.Count, value=1)

    return {"orderId": "1000", "order_key": order_key}


@lambda_handler_decorator
def custom_exception_handler(
    handler: Callable[[dict, LambdaContext], dict] = None,
    event: dict = None,
    context: LambdaContext = None,
) -> dict:
    """Utility to convert idempotency errors to api gateway response"""
    # noinspection PyBroadException
    try:
        return handler(event, context)
    except IdempotencyAlreadyInProgressError:
        message = {"message": "Payment in process, please try again later"}
        return {"statusCode": 400, "body": json.dumps(message)}
    except IdempotencyKeyError:
        message = {"message": "Request is missing an idempotent key"}
        return {"statusCode": 400, "body": json.dumps(message)}
    except IdempotencyPersistenceLayerError:
        message = {
            "message": "Failed to communicate to the idempotency persistent layer"
        }
        return {"statusCode": 503, "body": json.dumps(message)}
    except PaymentServiceError:
        message = {"message": "Payment failed"}
        return {"statusCode": 402, "body": json.dumps(message)}
    except Exception:
        message = {"message": "Something went wrong"}
        return {"statusCode": 503, "body": json.dumps(message)}


@custom_exception_handler
@tracer.capture_lambda_handler
@logger.inject_lambda_context
@metrics.log_metrics(capture_cold_start_metric=True)
@idempotent(config=config, persistence_store=persistence_layer)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    try:
        logger.debug("Function: %s#%s", context.function_name, context.function_version)
        event = APIGatewayProxyEvent(event)
        payment = create_payment(json.loads(event.body))
        return {"statusCode": 200, "body": json.dumps(payment)}
    except Exception as e:
        logger.exception(e)
        raise