REST API with DynamoDB

The rest-api-dynamodb project is a self-contained AWS CDK application with five REST routes backed by one DynamoDB table. It also creates an audit bucket whose read permission demonstrates role configuration. The handlers are grouped in one source module, while shared runtime helpers live in an AWS Lambda Layer.

Architecture

The stack contains one API Gateway REST API, five independent AWS Lambda Functions, one shared on-demand DynamoDB table with string partition key id, and one S3 audit bucket. lambdas/orders.py contains five decorated functions, and each function declares exactly one route. CDK creates a different entrypoint for each function even though the source file is shared:

• orders.list_orders — GET /orders
• orders.get_order — GET /orders/{id}
• orders.create_order — POST /orders
• orders.update_order — PUT /orders/{id}
• orders.delete_order — DELETE /orders/{id}

Grouping handlers by file does not group them into one Lambda. The five functions use the same orders AWS Lambda Layer, but remain independent resources with independent execution roles unless a role is selected explicitly. The POST handler selects api-role; the other four functions retain independent CDK-created execution roles.

Infrastructure

The complete rest_api_dynamodb/rest_api_dynamodb_stack.py is:

from aws_cdk import CfnOutput, RemovalPolicy, Stack
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_s3 as s3
from constructs import Construct
from lambda_api_decorators_cdk import LambdaApi, LambdaApiConfig


class RestApiDynamodbStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        table = dynamodb.Table(
            self,
            "Orders",
            table_name="Orders",
            partition_key=dynamodb.Attribute(
                name="id", type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY,
        )

        audit_bucket = s3.Bucket(
            self,
            "OrdersAudit",
            removal_policy=RemovalPolicy.DESTROY,
        )

        api_role = iam.Role(
            self,
            "ApiRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    "service-role/AWSLambdaBasicExecutionRole"
                )
            ],
        )
        audit_bucket.grant_read(api_role)

        config = LambdaApiConfig(
            default_runtime="python3.14",
            common_environment={"SERVICE": "orders"},
            environment_registry={
                "STAGE": {"STAGE": "dev"},
                "TABLE_NAME": {"TABLE_NAME": table.table_name},
            },
            role_registry={"api-role": api_role},
            dynamodb_table_registry={"orders": table},
        )

        api = LambdaApi(
            self,
            "Api",
            lambda_path="lambdas",
            layers_path="layers",
            config=config,
        )
        CfnOutput(self, "ApiUrl", value=api.api.url)

The table uses PAY_PER_REQUEST billing and RemovalPolicy.DESTROY. The ApiUrl output is the base URL used by the requests below.

Lambda handlers

The five functions live in one source module. Each function has exactly one HTTP route, its own DynamoDB grant and its own CDK-generated Lambda resource. They all import shared helpers from the Layer.

Function	Entrypoint	Method	Path
list_orders	orders.list_orders	GET	/orders
get_order	orders.get_order	GET	/orders/{id}
create_order	orders.create_order	POST	/orders
update_order	orders.update_order	PUT	/orders/{id}
delete_order	orders.delete_order	DELETE	/orders/{id}

lambdas/orders.py

from lambda_api_decorators import (
    DELETE,
    GET,
    POST,
    PUT,
    description,
    environment,
    grant_dynamodb,
    layer,
    memory_size,
    name,
    role,
    runtime,
    timeout,
)

from orders_shared import order_id, request_json, response, table


@GET("/orders")
@layer("orders")
@grant_dynamodb("orders", "read")
@environment("TABLE_NAME")
def list_orders(event, context):
    items = table().scan().get("Items", [])
    return response(200, items)


@GET("/orders/{id}")
@layer("orders")
@grant_dynamodb("orders", "read")
@environment("TABLE_NAME")
def get_order(event, context):
    item = table().get_item(Key={"id": order_id(event)}).get("Item")
    if item is None:
        return response(404, {"error": "Order not found"})
    return response(200, item)


@POST("/orders")
@layer("orders")
@grant_dynamodb("orders", "write")
@memory_size(1024)
@timeout(15)
@environment("STAGE", "TABLE_NAME")
@runtime("python3.12")
@role("api-role")
@description("Configured endpoint")
@name("configured-handler")
def create_order(event, context):
    order, error = request_json(event)
    if error:
        return error
    if "id" not in order:
        return response(400, {"error": "Missing required fields", "fields": ["id"]})
    table().put_item(Item=order, ConditionExpression="attribute_not_exists(id)")
    return response(201, order)


@PUT("/orders/{id}")
@layer("orders")
@grant_dynamodb("orders", "write")
@environment("TABLE_NAME")
def update_order(event, context):
    identifier = order_id(event)
    order, error = request_json(event)
    if error:
        return error
    existing = table().get_item(Key={"id": identifier}).get("Item")
    if existing is None:
        return response(404, {"error": "Order not found"})
    updated = {"id": identifier, **order}
    table().put_item(Item=updated)
    return response(200, updated)


@DELETE("/orders/{id}")
@layer("orders")
@grant_dynamodb("orders", "write")
@environment("TABLE_NAME")
def delete_order(event, context):
    deleted = table().delete_item(
        Key={"id": order_id(event)}, ReturnValues="ALL_OLD"
    ).get("Attributes")
    if deleted is None:
        return response(404, {"error": "Order not found"})
    return response(204, None)

create_order is the configured handler: it grants DynamoDB write access, uses 1024 MB and 15 seconds, selects STAGE and TABLE_NAME, overrides Python 3.14 with Python 3.12, selects api-role, and sets the description and function name.

AWS Lambda Layer

layers/orders/python/orders_shared.py

import json
import os

import boto3


JSON_HEADERS = {"Content-Type": "application/json"}
REQUIRED_FIELDS = ("customer", "total", "status")


def table():
    return boto3.resource("dynamodb").Table(os.environ["TABLE_NAME"])


def response(status_code, payload):
    return {
        "statusCode": status_code,
        "headers": JSON_HEADERS,
        "body": "" if status_code == 204 else json.dumps(payload),
    }


def request_json(event):
    try:
        body = json.loads(event.get("body") or "")
    except (TypeError, json.JSONDecodeError):
        return None, response(400, {"error": "Invalid JSON body"})
    if not isinstance(body, dict):
        return None, response(400, {"error": "JSON body must be an object"})
    return body, None


def order_id(event):
    return (event.get("pathParameters") or {}).get("id")


def missing_fields(order):
    return [field for field in REQUIRED_FIELDS if field not in order]

The Layer contains only shared helpers: it has no handlers, HTTP routes or infrastructure decorators. Its python directory is mounted as /opt/python by Lambda, and /opt/python is part of Python's import path. That is why the handlers can import orders_shared directly. layers_path="layers" defines the discovery directory, and @layer("orders") selects the discovered Layer for each function. A new Layer version is generated when its content changes.

Main files and tests

rest-api-dynamodb/
├── app.py
├── cdk.json
├── requirements-dev.txt
├── requirements.txt
├── rest_api_dynamodb/
│   ├── __init__.py
│   └── rest_api_dynamodb_stack.py
├── lambdas/
│   ├── __init__.py
│   ├── orders.py
│   └── requirements.txt
├── layers/
│   └── orders/
│       ├── requirements.txt
│       └── python/
│           └── orders_shared.py
└── tests/
    ├── __init__.py
    ├── test_handlers.py
    └── test_stack.py

tests/test_handlers.py imports lambdas.orders as the single handler module, exposes the Layer's Python directory for local tests, and uses a fake DynamoDB table to check listing, retrieval, creation, update, deletion, and invalid POST input. tests/test_stack.py uses AST and CDK assertions to check five handler functions, five distinct entrypoints, one route per function, one AWS::Lambda::LayerVersion, its association with all five Lambdas, imports from orders_shared, the five synthesized routes, and DynamoDB permissions. It also verifies that the named registries are present.

The repository-level pytest configuration uses --import-mode=importlib, which avoids test-module name collisions between the independent examples.

note

Docker must be running for the infrastructure tests and cdk synth, because CDK packages each Python Lambda with PythonFunction.

Project details

• dynamodb_table_registry makes the logical key orders resolvable; it does not grant IAM permissions. Each @grant_dynamodb declaration grants them explicitly.
• The stack creates an OrdersAudit S3 bucket and calls audit_bucket.grant_read(api_role), so api-role has read access to that bucket in addition to its basic Lambda execution policy. The handlers do not use the bucket; it is included to demonstrate a role with a concrete non-DynamoDB permission.
• read selects the library's DynamoDB read grant. write follows the published builder behavior and selects grant_read_write_data.
• @environment("STAGE", "TABLE_NAME") selects maps from environment_registry, which are merged with common_environment.
• POST overrides the default Python 3.14 runtime with @runtime("python3.12") and selects @role("api-role").
• The other Lambdas keep independent CDK-created roles. Sharing one role across handlers with grants accumulates the union of their permissions.
• Layer discovery identifies the available orders Layer; @layer("orders") selects it for a function. Sharing a Layer does not share execution roles or IAM permissions.
• Every handler represents exactly one HTTP route.
• RemovalPolicy.DESTROY is appropriate for this disposable example, but destroying the stack deletes the table and all its data.

Run the project

Run these commands from rest-api-dynamodb.

Linux/macOS

python3.14 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-dev.txt
pytest -q
cdk synth --quiet
cdk bootstrap
cdk deploy

Windows PowerShell

py -3.14 -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements-dev.txt
pytest -q
cdk synth --quiet
cdk bootstrap
cdk deploy

Windows CMD

py -3.14 -m venv .venv
.venv\Scripts\activate.bat
python -m pip install -r requirements-dev.txt
pytest -q
cdk synth --quiet
cdk bootstrap
cdk deploy

caution

Run cdk bootstrap and cdk deploy only with the intended AWS account and credentials configured.

Test the API

Set API_URL to the ApiUrl output from deployment. These requests use one consistent order with ID 123:

export API_URL="https://YOUR_REST_API_ID.execute-api.YOUR_REGION.amazonaws.com/prod"

curl -X POST "$API_URL/orders" \
  -H 'content-type: application/json' \
  -d '{"id":"123","customer":"Ada","total":10}'
curl "$API_URL/orders"
curl "$API_URL/orders/123"
curl -X PUT "$API_URL/orders/123" \
  -H 'content-type: application/json' \
  -d '{"customer":"Ada","total":12}'
curl -i -X DELETE "$API_URL/orders/123"

The same requests work in PowerShell or CMD after replacing $API_URL with the deployed URL, or by using the URL directly.

Clean up

cdk destroy

danger

The table uses RemovalPolicy.DESTROY. cdk destroy deletes the DynamoDB table and all of its data.

Complete source

See the complete rest-api-dynamodb directory for the runnable project and its README.