New serverless pattern - apigw-python-cdk-lambda-snapstart

apigw-python-cdk-lambda-snapstart/CarHandler/handler.py

@@ -0,0 +1,125 @@
+import base64
+import json
+import logging
+import os
+import uuid
+from decimal import Decimal
+from typing import Any, Dict, Optional
+
+import boto3
+from botocore.exceptions import ClientError
+
+logger = logging.getLogger()
+logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))
+
+table_name = os.environ["CAR_TABLE_NAME"]
+dynamodb = boto3.resource("dynamodb")
+table = dynamodb.Table(table_name)
+
+def _json_default(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        if value % 1 == 0:
+            return int(value)
+        return float(value)
+    raise TypeError(f"Object of type {type(value)} is not JSON serializable")
+
+def _response(status_code: int, body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    payload = "" if body is None else json.dumps(body, default=_json_default)
+    return {
+        "statusCode": status_code,
+        "headers": {"Content-Type": "application/json"},
+        "body": payload,
+    }
+
+def _parse_body(event: Dict[str, Any]) -> Dict[str, Any]:
+    raw = event.get("body") or "{}"
+    if event.get("isBase64Encoded", False):
+        raw = base64.b64decode(raw).decode("utf-8")
+    try:
+        body = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ValueError("Invalid JSON body") from exc
+
+    if not isinstance(body, dict):
+        raise ValueError("Request body must be a JSON object")
+
+    return body
+
+def _path_parts(path: str) -> list[str]:
+    if not path:
+        return []
+    return [part for part in path.strip("/").split("/") if part]
+
+def _create_car(event: Dict[str, Any]) -> Dict[str, Any]:
+    body = _parse_body(event)
+    car_id = str(uuid.uuid4())
+    car = {
+        "id": car_id,
+        "make": body.get("make"),
+        "model": body.get("model"),
+        "year": body.get("year"),
+        "color": body.get("color"),
+    }
+    table.put_item(Item=car)
+    return _response(201, car)
+
+def _get_car(car_id: str) -> Dict[str, Any]:
+    item = table.get_item(Key={"id": car_id}).get("Item")
+    if not item:
+        return _response(404, {"message": f"Car with id {car_id} not found"})
+    return _response(200, item)
+
+def _update_car(event: Dict[str, Any], car_id: str) -> Dict[str, Any]:
+    body = _parse_body(event)
+
+    existing = table.get_item(Key={"id": car_id}).get("Item")
+    if not existing:
+        return _response(404, {"message": f"Car with id {car_id} not found"})
+
+    updated = {
+        "id": car_id,
+        "make": body.get("make", existing.get("make")),
+        "model": body.get("model", existing.get("model")),
+        "year": body.get("year", existing.get("year")),
+        "color": body.get("color", existing.get("color")),
+    }
+    table.put_item(Item=updated)
+    return _response(200, updated)
+
+
+def _delete_car(car_id: str) -> Dict[str, Any]:
+    try:
+        table.delete_item(
+            Key={"id": car_id},
+            ConditionExpression="attribute_exists(id)",
+        )
+    except ClientError as exc:
+        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
+            return _response(404, {"message": f"Car with id {car_id} not found"})
+        raise
+
+    return _response(204)
+
+def handler(event: Dict[str, Any], _: Any) -> Dict[str, Any]:
+    method = (event.get("httpMethod") or "").upper()
+    path = event.get("path") or ""
+    parts = _path_parts(path)
+
+    logger.info(json.dumps({"method": method, "path": path}))
+
+    try:
+        if method == "POST" and parts == ["cars"]:
+            return _create_car(event)
+
+        if len(parts) == 2 and parts[0] == "cars":
+            car_id = parts[1]
+            if method == "GET":
+                return _get_car(car_id)
+            if method == "PUT":
+                return _update_car(event, car_id)
+            if method == "DELETE":
+                return _delete_car(car_id)
+
+        return _response(404, {"message": "Route not found"})
+    except ValueError as exc:
+        return _response(400, {"message": str(exc)})

apigw-python-cdk-lambda-snapstart/README.md

@@ -0,0 +1,49 @@
+# API Gateway + Lambda SnapStart + DynamoDB (Python CDK)
+
+This pattern demonstrates how to create a REST API using API Gateway, AWS Lambda and DynamoDB.
+It's built with [Python 3.12](https://www.python.org/downloads/release/python-3128/), together with
+[AWS Cloud Development Kit (CDK)](https://docs.aws.amazon.com/cdk/v2/guide/work-with-cdk-python.html) as the Infrastructure as Code solution. This pattern also implements the usage of [AWS Lambda SnapStart](https://docs.aws.amazon.com/lambda/latest/dg/snapstart.html)
+to improve initialization performance of the Lambda, in case that there are libraries requiring more time to load into memory.
+
+## Architecture
+
+- API Gateway REST API (`prod` stage)
+- AWS Lambda (Python 3.12)
+- Lambda SnapStart enabled on published versions
+- Lambda `live` alias integrated with API Gateway
+- DynamoDB table with partition key `id`
+
+## Endpoints
+
+- `POST /cars`
+- `GET /cars/{carId}`
+- `PUT /cars/{carId}`
+- `DELETE /cars/{carId}`
+
+## Requirements
+
+- Python 3.12+
+- AWS CDK v2
+- AWS credentials configured
+
+## Deploy
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+cdk bootstrap
+cdk deploy
+```
+
+## Test
+
+Get endpoint URL from stack outputs (`CarEndpoint`), then run:
+
+```bash
+ENDPOINT="<put-the-CarEndpoint-output-URL-here>"
+
+curl --location --request POST "$ENDPOINT/cars" \
+  --header 'Content-Type: application/json' \
+  --data-raw '{"make":"Porsche","model":"992","year":"2022","color":"White"}'
+```

apigw-python-cdk-lambda-snapstart/app.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+from aws_cdk import (
+    App, CfnOutput,
+    Duration,
+    Stack,
+    RemovalPolicy,
+    aws_apigateway as apigw,
+    aws_dynamodb as dynamodb,
+    aws_lambda as _lambda,
+)
+from constructs import Construct
+
+class CarStoreStack(Stack):
+    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
+        super().__init__(scope, construct_id, **kwargs)
+
+        car_table = dynamodb.Table(
+            self,
+            "CarTable",
+            partition_key=dynamodb.Attribute(name="id", type=dynamodb.AttributeType.STRING),
+            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
+            removal_policy=RemovalPolicy.DESTROY,
+        )
+
+        car_function = _lambda.Function(
+            self,
+            "CarStoreFunction",
+            runtime=_lambda.Runtime.PYTHON_3_12,
+            handler="handler.handler",
+            code=_lambda.Code.from_asset("CarHandler/"),
+            timeout=Duration.seconds(10),
+            snap_start=_lambda.SnapStartConf.ON_PUBLISHED_VERSIONS,
+            memory_size=256,
+            environment={
+                "CAR_TABLE_NAME": car_table.table_name,
+                "LOG_LEVEL": "INFO",
+            }
+        )
+        car_table.grant_read_write_data(car_function)
+
+        live_alias = _lambda.Alias(
+            self,
+            "CarStoreLiveAlias",
+            alias_name="live",
+            version=car_function.current_version,
+        )
+
+        car_api = apigw.RestApi(
+            self,
+            "CarStoreApi",
+            deploy_options=apigw.StageOptions(stage_name="prod"),
+        )
+
+        integration = apigw.LambdaIntegration(live_alias, proxy=True)
+        car_api.root.add_method("ANY", integration)
+        car_api.root.add_resource("{proxy+}").add_method("ANY", integration)
+
+        CfnOutput(
+            self,
+            "CarEndpoint",
+            description="API Gateway Car Endpoint",
+            value=car_api.url,
+        )
+        CfnOutput(self, "CarTableName", value=car_table.table_name)
+
+app = App()
+CarStoreStack(app, "CarStoreStack")
+app.synth()

apigw-python-cdk-lambda-snapstart/cdk.json

@@ -0,0 +1,15 @@
+{
+  "app": "../.venv/bin/python app.py",
+  "watch": {
+    "include": [
+      "**"
+    ],
+    "exclude": [
+      "README.md",
+      "cdk*.json",
+      "requirements*.txt",
+      "**/__pycache__",
+      "tests"
+    ]
+  }
+}

apigw-python-cdk-lambda-snapstart/example-pattern.json

@@ -0,0 +1,73 @@
+{
+  "title": "API Gateway with Lambda SnapStart and DynamoDB using Python CDK",
+  "description": "This pattern demonstrates how to create a REST API using API Gateway, AWS Lambda with SnapStart, and DynamoDB. Built with Python 3.12 and AWS CDK, it implements Lambda SnapStart to improve initialization performance for faster cold starts.",
+  "language": "Python",
+  "level": "200",
+  "framework": "AWS CDK",
+  "introBox": {
+    "headline": "How it works",
+    "text": [
+      "This pattern creates a REST API for managing car records using API Gateway and Lambda with SnapStart enabled.",
+      "The Lambda function is Python 3.12 based and includes a live alias that's integrated with API Gateway for seamless deployments.",
+      "Lambda SnapStart persists the initialized state of the Lambda runtime, significantly reducing cold start times for function initialization.",
+      "DynamoDB stores car records with a partition key of 'id', providing a scalable NoSQL database backend for the REST API."
+    ]
+  },
+  "gitHub": {
+    "template": {
+      "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/apigw-python-cdk-lambda-snapstart",
+      "templateURL": "serverless-patterns/apigw-python-cdk-lambda-snapstart",
+      "projectFolder": "apigw-python-cdk-lambda-snapstart",
+      "templateFile": "apigw-python-cdk-lambda-snapstart/app.py"
+    }
+  },
+  "resources": {
+    "bullets": [
+      {
+        "text": "AWS Lambda SnapStart",
+        "link": "https://docs.aws.amazon.com/lambda/latest/dg/snapstart.html"
+      },
+      {
+        "text": "API Gateway REST API",
+        "link": "https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-rest-api.html"
+      },
+      {
+        "text": "AWS CDK Python Reference",
+        "link": "https://docs.aws.amazon.com/cdk/v2/guide/work-with-cdk-python.html"
+      }
+    ]
+  },
+  "deploy": {
+    "text": [
+      "python3 -m venv .venv",
+      "source .venv/bin/activate",
+      "pip install -r requirements.txt",
+      "cdk bootstrap",
+      "cdk deploy"
+    ]
+  },
+  "testing": {
+    "text": [
+      "Get the CarEndpoint from stack outputs, then test the endpoint to create a new car record:",
+      "curl --location --request POST \"$ENDPOINT/cars\" --header 'Content-Type: application/json' --data-raw '{\"make\":\"Porsche\",\"model\":\"992\",\"year\":\"2022\",\"color\":\"White\"}'",
+      "Change the endpoint and HTTP method to test other operations:",
+      "GET /cars/{carId} - Retrieve a car",
+      "PUT /cars/{carId} - Update a car",
+      "DELETE /cars/{carId} - Delete a car"
+    ]
+  },
+  "cleanup": {
+    "text": [
+      "Delete the stack: <code>cdk destroy</code>."
+    ]
+  },
+  "authors": [
+    {
+      "name": "Matia Rasetina",
+      "image": "https://media.licdn.com/dms/image/v2/C4D03AQEpZLzvymfGyA/profile-displayphoto-shrink_800_800/profile-displayphoto-shrink_800_800/0/1612951581132?e=1772668800&v=beta&t=m8AkoSUFICMRk5-Gd0hEAji0N4gFSfFGuv4lbBuXcJY",
+      "bio": "Senior Software Engineer @ Elixirr Digital",
+      "linkedin": "https://www.linkedin.com/in/matiarasetina/",
+      "twitter": ""
+    }
+  ]
+}

apigw-python-cdk-lambda-snapstart/requirements.txt

@@ -0,0 +1,2 @@
+aws-cdk-lib>=2.170.0
+constructs>=10.0.0,<11.0.0