08 - Validator

app-validator.py

import json
import os
import time
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.metrics import MetricUnit
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.typing import LambdaContext
from aws_lambda_powertools.utilities.validation import validator

logger = Logger()
tracer = Tracer()
metrics = Metrics()
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"},
    },
}


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


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)}


@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"})


@tracer.capture_method
def create_payment(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
    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"}


@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=INPUT, outbound_schema=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 = create_payment(payment_request)
    return build_response(body=payment)