app - modular

app.py

import os
from typing import Any, Callable, Dict

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.logging import correlation_paths
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 DynamoDBPersistenceLayer, IdempotencyConfig, idempotent
from aws_lambda_powertools.utilities.idempotency.exceptions import (
    IdempotencyAlreadyInProgressError,
    IdempotencyKeyError,
    IdempotencyPersistenceLayerError,
)
from aws_lambda_powertools.utilities.parameters import get_parameter
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.validation import validator

from .clients import PaymentService
from .exceptions import PaymentServiceError
from . import schemas
from .utils import build_response

logger = Logger()
tracer = Tracer()
metrics = Metrics()
credentials = get_parameter("payment_credentials", transform="json", decrypt=True)
assert isinstance(credentials, dict)
payment_client = PaymentService(**credentials)


@lambda_handler_decorator(trace_execution=True)
def exception_handler(handler: Callable[[Dict, LambdaContext], Dict], event: Dict, context: LambdaContext) -> Dict:
    """Utility to convert errors to api gateway response"""
    try:
        proxy_event = APIGatewayProxyEvent(event)
        assert proxy_event.get_header_value("x-api-key") == os.environ["SOME_STATIC_KEY"]
        return handler(event, context)
    except Exception as ex:
        tracer.put_annotation("ERROR_TYPE", type(ex).__name__)
        logger.exception("Failed to perform handler")
        if isinstance(ex, IdempotencyAlreadyInProgressError):
            return build_response(400, {"message": "Transaction in progress, please try again later"})
        if isinstance(ex, IdempotencyKeyError):
            return build_response(400, {"message": "Request is missing an idempotent key"})
        if isinstance(ex, IdempotencyPersistenceLayerError):
            return build_response(503, {"message": "Failure with idempotency persistent layer"})
        if isinstance(ex, PaymentServiceError):
            return build_response(402, {"message": "Payment failed"})
        return build_response(503, {"message": "Something went wrong"})


@exception_handler
@metrics.log_metrics(capture_cold_start_metric=True)
@tracer.capture_lambda_handler
@logger.inject_lambda_context(log_event=True, correlation_id_path=correlation_paths.API_GATEWAY_REST)
@validator(inbound_schema=schemas.INPUT, outbound_schema=schemas.OUTPUT, envelope="powertools_json(body)")
@idempotent(
    config=IdempotencyConfig(
        event_key_jmespath="order_key",
        raise_on_no_idempotency_key=True,
        use_local_cache=True,
        local_cache_max_items=512,
        expires_after_seconds=24 * 60 * 60,
    ),
    persistence_store=DynamoDBPersistenceLayer(table_name=os.environ["IDEMPOTENCY_TABLE_NAME"]),
)
def lambda_handler(payment_request: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
    logger.debug("Function: %s#%s", context.function_name, context.function_version)
    payment = payment_client.create(payment_request)
    return build_response(body=payment)

clients.py

import time
from typing import Dict

from aws_lambda_powertools import Tracer, Logger, Metrics
from aws_lambda_powertools.metrics import MetricUnit

from .exceptions import PaymentServiceError

tracer = Tracer()
logger = Logger(child=True)
metrics = Metrics()


class PaymentService:
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret

    @tracer.capture_method
    def create(self, payment: Dict[str, str]) -> Dict[str, str]:
        order_key = payment["order_key"]
        logger.info(f"Order key: %s", order_key)

        # Simulate failure
        if "FAILURE" in order_key:
            tracer.put_annotation("PAYMENT_STATUS", "ERROR")
            metrics.add_metric(name="PaymentFailure", unit=MetricUnit.Count, value=1)
            raise PaymentServiceError("Failed to make payment")

        # Simulate a slow transaction
        logger.debug("Mock call using api_key: %s", self.api_key)
        time.sleep(2)
        tracer.put_annotation(key="ORDER_KEY", value=order_key)
        tracer.put_annotation(key="PAYMENT_STATUS", value="SUCCESS")
        metrics.add_metric(name="PaymentSuccessful", unit=MetricUnit.Count, value=1)

        return {"order_id": "1111111", "status": "SUCCESS"}

exceptions.py

class PaymentServiceError(Exception):
    """Payment service error"""

schemas.py

INPUT = {
    "type": "object",
    "title": "Payment request",
    "examples": [{"order_key": "FB92B90B-763E-4FE2-BD1F-595A8DFAC75C", "amount": 100}],
    "required": ["order_key", "amount"],
    "properties": {
        "order_key": {
            "type": "string",
            "title": "Order key",
            "examples": ["FB92B90B-763E-4FE2-BD1F-595A8DFAC75C"],
            "maxLength": 36,
        },
        "amount": {
            "type": "integer",
            "title": "Amount in cents",
            "examples": [100],
        },
        "currency_code": {
            "type": "string",
            "title": "Currency Code",
            "examples": ["USD"],
            "minLength": 3,
            "maxLength": 3,
        },
    },
}
OUTPUT = {
    "type": "object",
    "required": ["statusCode", "body"],
    "properties": {
        "statusCode": {"type": "integer"},
        "body": {"type": "string"},
    },
}

utils.py

import json
from typing import Dict, Any


def build_response(status_code: int = 200, body: Dict[str, Any] = None) -> Dict[str, Any]:
    """Build an api gateway response"""
    return {"statusCode": status_code, "headers": {"Content-Type": "application/json"}, "body": json.dumps(body)}