lambda_handler.py

# Built-ins
from os import getenv
from json import dumps
from typing import Any, Dict
from http import HTTPStatus
# Third party
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.parser import event_parser, BaseModel
import boto3

class UserInsertionEvent(BaseModel):
    firstname: str
    lastname: str
    created_at: str

@event_parser(model=UserInsertionEvent)
def handler(event: UserInsertionEvent, context: LambdaContext) -> Dict[str, Any]:
    client = boto3.client("sns")
    response = client.publish(
        TopicArn=getenv("NOTIFICATION_TOPIC_ARN"),
        Message=f"A new user was added at {event.created_at}: {event.firstname} {event.lastname}"
    )
    return {
        "lambda_request_id": context.aws_request_id,
        "lambda_arn": context.invoked_function_arn,
        "status_code": HTTPStatus.OK.value,
        "event": event.json(),
        "response": response
    }