DynamoDB Pulumi: TypeScript and Python Table Examples

Create a DynamoDB table with Pulumi's aws.dynamodb.Table in TypeScript or pulumi_aws.dynamodb.Table in Python. The current Pulumi AWS provider major is v7, and it supports on-demand capacity, indexes, backups, deletion protection, warm throughput and global replicas. Pulumi provisions infrastructure; application reads and writes use an AWS SDK separately.

Why use Pulumi for DynamoDB?

Pulumi lets your infrastructure use the same language, package management and abstractions as your application. You can export the table name, pass it to Lambda and construct scoped IAM policies from the table ARN. Resources can live in imported modules; they do not all have to be declared directly in index.ts.

Terraform provides similar infrastructure management through HCL. Choose based on the team's existing state management, review process and language preferences. Pulumi's TypeScript does not eliminate the need to understand table replacement, IAM or DynamoDB access patterns. SST adds application-oriented components and linking on top of Pulumi-based infrastructure.

The examples use the Pulumi AWS Table reference. Pulumi AWS v7 and Terraform AWS v6 are separate package version lines; do not assume their major numbers should match.

Step 1: initialize a TypeScript stack

Install the Pulumi CLI, Node.js and AWS CLI v2. Authenticate with your chosen Pulumi state backend and configure an AWS development profile:

export AWS_PROFILE=development
aws sts get-caller-identity
pulumi version
mkdir pulumi-dynamodb-api
cd pulumi-dynamodb-api
pulumi login
pulumi new aws-typescript
pulumi config set aws:region us-east-1
npm install @pulumi/aws@7
export AWS_PROFILE=development
aws sts get-caller-identity
pulumi version
mkdir pulumi-dynamodb-api
cd pulumi-dynamodb-api
pulumi login
pulumi new aws-typescript
pulumi config set aws:region us-east-1
npm install @pulumi/aws@7

Choose a project name and a dev stack. Keep the package lock file and review provider upgrades before running them against persistent data.

These screenshots illustrate the workflow; CLI versions and prompts have changed. Replace any generated sample bucket with the table code below.

Step 2: provision the users table

Create dynamo/users.ts. This complete definition keeps the original composite key, adds useful secondary indexes and enables backups:

import * as aws from "@pulumi/aws"

export const Users = new aws.dynamodb.Table("users", {
  billingMode: "PAY_PER_REQUEST",
  hashKey: "id",
  rangeKey: "createdAt",
  attributes: [
    { name: "id", type: "S" },
    { name: "createdAt", type: "N" },
    { name: "email", type: "S" },
    { name: "name", type: "S" },
  ],
  globalSecondaryIndexes: [{
    name: "ByEmail",
    hashKey: "email",
    rangeKey: "createdAt",
    projectionType: "ALL",
  }],
  localSecondaryIndexes: [{
    name: "ByName",
    rangeKey: "name",
    projectionType: "INCLUDE",
    nonKeyAttributes: ["email", "age"],
  }],
  ttl: { enabled: true, attributeName: "deleteAt" },
  pointInTimeRecovery: { enabled: true, recoveryPeriodInDays: 14 },
  deletionProtectionEnabled: true,
  streamEnabled: true,
  streamViewType: "NEW_AND_OLD_IMAGES",
  tags: { Application: "users" },
}, { protect: true, retainOnDelete: true })
import * as aws from "@pulumi/aws"

export const Users = new aws.dynamodb.Table("users", {
  billingMode: "PAY_PER_REQUEST",
  hashKey: "id",
  rangeKey: "createdAt",
  attributes: [
    { name: "id", type: "S" },
    { name: "createdAt", type: "N" },
    { name: "email", type: "S" },
    { name: "name", type: "S" },
  ],
  globalSecondaryIndexes: [{
    name: "ByEmail",
    hashKey: "email",
    rangeKey: "createdAt",
    projectionType: "ALL",
  }],
  localSecondaryIndexes: [{
    name: "ByName",
    rangeKey: "name",
    projectionType: "INCLUDE",
    nonKeyAttributes: ["email", "age"],
  }],
  ttl: { enabled: true, attributeName: "deleteAt" },
  pointInTimeRecovery: { enabled: true, recoveryPeriodInDays: 14 },
  deletionProtectionEnabled: true,
  streamEnabled: true,
  streamViewType: "NEW_AND_OLD_IMAGES",
  tags: { Application: "users" },
}, { protect: true, retainOnDelete: true })

Import it from index.ts:

import { Users } from "./dynamo/users"
export const usersTableName = Users.name
export const usersTableArn = Users.arn
import { Users } from "./dynamo/users"
export const usersTableName = Users.name
export const usersTableArn = Users.arn

The logical name users is not necessarily the physical table name: Pulumi can generate a unique suffix. Export Users.name rather than assuming a literal name. Values such as name and arn are Outputs; pass them to other resources or use pulumi.interpolate, rather than calling .get() during deployment.

pulumi preview --stack dev
pulumi up --stack dev
pulumi stack output usersTableName
pulumi preview --stack dev
pulumi up --stack dev
pulumi stack output usersTableName

The current preview includes the protections and settings above. Check changes before approving pulumi up, particularly anything marked replace or delete.

The same table in Python

Use this in __main__.py of a separate pulumi new aws-python project. Install a reviewed pulumi-aws v7 release in its virtual environment and set the AWS Region as above. This is an alternative implementation, not a second owner for the TypeScript table.

import pulumi
import pulumi_aws as aws

users = aws.dynamodb.Table(
    "users",
    billing_mode="PAY_PER_REQUEST",
    hash_key="id",
    range_key="createdAt",
    attributes=[
        aws.dynamodb.TableAttributeArgs(name="id", type="S"),
        aws.dynamodb.TableAttributeArgs(name="createdAt", type="N"),
        aws.dynamodb.TableAttributeArgs(name="email", type="S"),
        aws.dynamodb.TableAttributeArgs(name="name", type="S"),
    ],
    global_secondary_indexes=[aws.dynamodb.TableGlobalSecondaryIndexArgs(
        name="ByEmail", hash_key="email", range_key="createdAt", projection_type="ALL",
    )],
    local_secondary_indexes=[aws.dynamodb.TableLocalSecondaryIndexArgs(
        name="ByName", range_key="name", projection_type="INCLUDE",
        non_key_attributes=["email", "age"],
    )],
    ttl=aws.dynamodb.TableTtlArgs(enabled=True, attribute_name="deleteAt"),
    point_in_time_recovery=aws.dynamodb.TablePointInTimeRecoveryArgs(
        enabled=True, recovery_period_in_days=14,
    ),
    deletion_protection_enabled=True,
    stream_enabled=True,
    stream_view_type="NEW_AND_OLD_IMAGES",
    tags={"Application": "users"},
    opts=pulumi.ResourceOptions(protect=True, retain_on_delete=True),
)

pulumi.export("usersTableName", users.name)
pulumi.export("usersTableArn", users.arn)
import pulumi
import pulumi_aws as aws

users = aws.dynamodb.Table(
    "users",
    billing_mode="PAY_PER_REQUEST",
    hash_key="id",
    range_key="createdAt",
    attributes=[
        aws.dynamodb.TableAttributeArgs(name="id", type="S"),
        aws.dynamodb.TableAttributeArgs(name="createdAt", type="N"),
        aws.dynamodb.TableAttributeArgs(name="email", type="S"),
        aws.dynamodb.TableAttributeArgs(name="name", type="S"),
    ],
    global_secondary_indexes=[aws.dynamodb.TableGlobalSecondaryIndexArgs(
        name="ByEmail", hash_key="email", range_key="createdAt", projection_type="ALL",
    )],
    local_secondary_indexes=[aws.dynamodb.TableLocalSecondaryIndexArgs(
        name="ByName", range_key="name", projection_type="INCLUDE",
        non_key_attributes=["email", "age"],
    )],
    ttl=aws.dynamodb.TableTtlArgs(enabled=True, attribute_name="deleteAt"),
    point_in_time_recovery=aws.dynamodb.TablePointInTimeRecoveryArgs(
        enabled=True, recovery_period_in_days=14,
    ),
    deletion_protection_enabled=True,
    stream_enabled=True,
    stream_view_type="NEW_AND_OLD_IMAGES",
    tags={"Application": "users"},
    opts=pulumi.ResourceOptions(protect=True, retain_on_delete=True),
)

pulumi.export("usersTableName", users.name)
pulumi.export("usersTableArn", users.arn)

Python uses snake_case argument names and typed Args objects. TypeScript uses camelCase. Both definitions express the same DynamoDB settings; the application can use boto3 or SDK v3 independently of the infrastructure language.

Table settings and index behavior

TTL and point-in-time recovery

deleteAt is a Number containing Unix epoch seconds. JavaScript's Date.now() returns milliseconds, so divide by 1000 before adding a duration. TTL cleanup normally happens within a few days after expiry, not exactly at the expiry timestamp. Expired items remain readable until deleted unless the application filters or rejects them.

PITR supports a recovery window of 1 to 35 days, with 35 as the default. The example explicitly chooses 14. A shorter window does not reduce the PITR price, and restoration creates a new table rather than rolling the existing table backward. Test the restore and application cutover procedures.

Encryption

DynamoDB always encrypts data at rest. Omitting serverSideEncryption uses an AWS owned key. serverSideEncryption: { enabled: true } selects the AWS managed DynamoDB key, not the AWS owned key. Add kmsKeyArn to that object to choose a customer managed key.

Retain customer managed keys with the data and review their key policies. A table retained by Pulumi can still become unavailable if its encryption key is disabled or removed.

LSIs and GSIs

Only attributes used by table or index keys belong in attributes. The example's age and deleteAt are ordinary item fields and therefore absent. Key attributes are automatically projected into indexes; do not repeat them in nonKeyAttributes.

The ByName LSI shares the base table's id partition key and provides an alternate sort key. LSIs must be created with the table. Adding or changing one later requires replacement, so use a new-table migration rather than accepting a destructive preview.

The ByEmail GSI has a separate partition key. GSIs can be added later, but backfill must complete before the application relies on them. Their reads are eventually consistent. Indexes add storage and write costs; ALL projects every field, while INCLUDE narrows that copy.

On-demand caps, warm throughput and provisioned scaling

For a best-effort on-demand request-rate cap, add:

onDemandThroughput: {
  maxReadRequestUnits: 20000,
  maxWriteRequestUnits: 8000,
},
onDemandThroughput: {
  maxReadRequestUnits: 20000,
  maxWriteRequestUnits: 8000,
},

For planned pre-warming, add this separate optional setting:

warmThroughput: {
  readUnitsPerSecond: 20000,
  writeUnitsPerSecond: 8000,
},
warmThroughput: {
  readUnitsPerSecond: 20000,
  writeUnitsPerSecond: 8000,
},

Maximum throughput can throttle requests; it is not a hard monthly budget. Warm throughput prepares immediate capacity and proactive increases can incur a one-time charge. Do not configure it casually on a small development table.

For provisioned mode, set billingMode: "PROVISIONED", specify numeric readCapacity and writeCapacity, remove on-demand caps and set capacities on each GSI. Application Auto Scaling uses separate targets and policies. When it owns capacity, use Pulumi resource options such as ignoreChanges: ["readCapacity", "writeCapacity"] so updates do not continually restore stale capacity settings. Review GSI capacity drift separately.

Global-table replicas

For an eventually consistent global table, add a replica to the TypeScript table definition; Streams are already enabled:

replicas: [{
  regionName: "us-west-2",
  pointInTimeRecovery: true,
  deletionProtectionEnabled: true,
  propagateTags: true,
}],
replicas: [{
  regionName: "us-west-2",
  pointInTimeRecovery: true,
  deletionProtectionEnabled: true,
  propagateTags: true,
}],

Configure replica backups and deletion protection explicitly. Do not manage the same replica both inline and through a standalone replica resource. The table and index key schemas must remain consistent across Regions.

The provider also exposes consistencyMode: "STRONG" on replicas and globalTableWitness: { regionName: "us-east-2" }. These support MRSC creation with three Regions: three data replicas, or two plus a witness. The example table cannot simply switch to MRSC because it has TTL and an LSI, both unsupported by MRSC. Transactions are also unsupported. Plan a separate empty table and review the global tables guide.

Step 3: implement the user API with SDK v3

Keep application code separate from Pulumi code. Do not use the old aws.sdk.DynamoDB.DocumentClient helper. Create lambda/index.mjs and install its own SDK dependencies:

mkdir lambda
cd lambda
npm init -y
npm install @aws-sdk/client-dynamodb @aws-sdk/lib-dynamodb
cd ..
mkdir lambda
cd lambda
npm init -y
npm install @aws-sdk/client-dynamodb @aws-sdk/lib-dynamodb
cd ..

This handler serves all four original operations. Both key components travel in update and delete paths. The example is for a disposable demonstration dataset: its five-day TTL is deliberate, not an appropriate default for permanent user accounts.

import { randomUUID } from "node:crypto"
import { DynamoDBClient } from "@aws-sdk/client-dynamodb"
import {
  DynamoDBDocumentClient, PutCommand, UpdateCommand, DeleteCommand, ScanCommand,
} from "@aws-sdk/lib-dynamodb"

const db = DynamoDBDocumentClient.from(new DynamoDBClient({}))
const TableName = process.env.TABLE_NAME
const json = (statusCode, value) => ({ statusCode, body: JSON.stringify(value) })

export const handler = async (event) => {
  const method = event.requestContext.http.method
  let body = {}
  if (event.body) {
    try {
      body = JSON.parse(event.isBase64Encoded
        ? Buffer.from(event.body, "base64").toString("utf8") : event.body)
      if (!body || typeof body !== "object" || Array.isArray(body)) throw new Error("Invalid body")
    } catch {
      return json(400, { message: "Body must be a JSON object" })
    }
  }

  if (method === "POST") {
    if (typeof body.name !== "string" || !body.name ||
        typeof body.email !== "string" || !body.email ||
        !Number.isInteger(body.age) || body.age < 0) {
      return json(400, { message: "name, email and a nonnegative integer age are required" })
    }
    const item = {
      id: randomUUID(), createdAt: Date.now(),
      name: body.name, email: body.email, age: body.age,
      deleteAt: Math.floor(Date.now() / 1000) + 5 * 24 * 60 * 60,
    }
    await db.send(new PutCommand({ TableName, Item: item }))
    return json(201, item)
  }

  if (method === "GET") {
    let ExclusiveStartKey
    const token = event.queryStringParameters?.nextToken
    if (token) {
      try {
        const key = JSON.parse(Buffer.from(token, "base64url").toString("utf8"))
        if (typeof key?.id !== "string" || !key.id || !Number.isSafeInteger(key.createdAt)) {
          throw new Error("Invalid key")
        }
        ExclusiveStartKey = { id: key.id, createdAt: key.createdAt }
      } catch {
        return json(400, { message: "Invalid nextToken" })
      }
    }
    const result = await db.send(new ScanCommand({ TableName, Limit: 100, ExclusiveStartKey }))
    return json(200, {
      items: result.Items ?? [],
      nextToken: result.LastEvaluatedKey
        ? Buffer.from(JSON.stringify(result.LastEvaluatedKey)).toString("base64url") : null,
    })
  }

  const id = event.pathParameters?.id
  const createdAt = Number(event.pathParameters?.createdAt)
  if (!id || !Number.isSafeInteger(createdAt)) return json(400, { message: "Invalid key" })
  const Key = { id, createdAt }

  if (method === "DELETE") {
    await db.send(new DeleteCommand({ TableName, Key }))
    return { statusCode: 204, body: "" }
  }

  if (method === "PATCH") {
    const fields = ["name", "email", "age"].filter((field) => Object.hasOwn(body, field))
    if (!fields.length) return json(400, { message: "No editable fields supplied" })
    if (fields.some((field) => field === "age"
      ? !Number.isInteger(body.age) || body.age < 0
      : typeof body[field] !== "string" || !body[field])) {
      return json(400, { message: "Invalid field value" })
    }
    try {
      const result = await db.send(new UpdateCommand({
        TableName, Key,
        ConditionExpression: "attribute_exists(id)",
        UpdateExpression: "SET " + fields.map((field) => `#${field} = :${field}`).join(", "),
        ExpressionAttributeNames: Object.fromEntries(fields.map((field) => [`#${field}`, field])),
        ExpressionAttributeValues: Object.fromEntries(fields.map((field) => [`:${field}`, body[field]])),
        ReturnValues: "ALL_NEW",
      }))
      return json(200, result.Attributes)
    } catch (error) {
      if (error.name === "ConditionalCheckFailedException") return json(404, { message: "Not found" })
      throw error
    }
  }
  return json(405, { message: "Method not allowed" })
}
import { randomUUID } from "node:crypto"
import { DynamoDBClient } from "@aws-sdk/client-dynamodb"
import {
  DynamoDBDocumentClient, PutCommand, UpdateCommand, DeleteCommand, ScanCommand,
} from "@aws-sdk/lib-dynamodb"

const db = DynamoDBDocumentClient.from(new DynamoDBClient({}))
const TableName = process.env.TABLE_NAME
const json = (statusCode, value) => ({ statusCode, body: JSON.stringify(value) })

export const handler = async (event) => {
  const method = event.requestContext.http.method
  let body = {}
  if (event.body) {
    try {
      body = JSON.parse(event.isBase64Encoded
        ? Buffer.from(event.body, "base64").toString("utf8") : event.body)
      if (!body || typeof body !== "object" || Array.isArray(body)) throw new Error("Invalid body")
    } catch {
      return json(400, { message: "Body must be a JSON object" })
    }
  }

  if (method === "POST") {
    if (typeof body.name !== "string" || !body.name ||
        typeof body.email !== "string" || !body.email ||
        !Number.isInteger(body.age) || body.age < 0) {
      return json(400, { message: "name, email and a nonnegative integer age are required" })
    }
    const item = {
      id: randomUUID(), createdAt: Date.now(),
      name: body.name, email: body.email, age: body.age,
      deleteAt: Math.floor(Date.now() / 1000) + 5 * 24 * 60 * 60,
    }
    await db.send(new PutCommand({ TableName, Item: item }))
    return json(201, item)
  }

  if (method === "GET") {
    let ExclusiveStartKey
    const token = event.queryStringParameters?.nextToken
    if (token) {
      try {
        const key = JSON.parse(Buffer.from(token, "base64url").toString("utf8"))
        if (typeof key?.id !== "string" || !key.id || !Number.isSafeInteger(key.createdAt)) {
          throw new Error("Invalid key")
        }
        ExclusiveStartKey = { id: key.id, createdAt: key.createdAt }
      } catch {
        return json(400, { message: "Invalid nextToken" })
      }
    }
    const result = await db.send(new ScanCommand({ TableName, Limit: 100, ExclusiveStartKey }))
    return json(200, {
      items: result.Items ?? [],
      nextToken: result.LastEvaluatedKey
        ? Buffer.from(JSON.stringify(result.LastEvaluatedKey)).toString("base64url") : null,
    })
  }

  const id = event.pathParameters?.id
  const createdAt = Number(event.pathParameters?.createdAt)
  if (!id || !Number.isSafeInteger(createdAt)) return json(400, { message: "Invalid key" })
  const Key = { id, createdAt }

  if (method === "DELETE") {
    await db.send(new DeleteCommand({ TableName, Key }))
    return { statusCode: 204, body: "" }
  }

  if (method === "PATCH") {
    const fields = ["name", "email", "age"].filter((field) => Object.hasOwn(body, field))
    if (!fields.length) return json(400, { message: "No editable fields supplied" })
    if (fields.some((field) => field === "age"
      ? !Number.isInteger(body.age) || body.age < 0
      : typeof body[field] !== "string" || !body[field])) {
      return json(400, { message: "Invalid field value" })
    }
    try {
      const result = await db.send(new UpdateCommand({
        TableName, Key,
        ConditionExpression: "attribute_exists(id)",
        UpdateExpression: "SET " + fields.map((field) => `#${field} = :${field}`).join(", "),
        ExpressionAttributeNames: Object.fromEntries(fields.map((field) => [`#${field}`, field])),
        ExpressionAttributeValues: Object.fromEntries(fields.map((field) => [`:${field}`, body[field]])),
        ReturnValues: "ALL_NEW",
      }))
      return json(200, result.Attributes)
    } catch (error) {
      if (error.name === "ConditionalCheckFailedException") return json(404, { message: "Not found" })
      throw error
    }
  }
  return json(405, { message: "Method not allowed" })
}

The PATCH code checks whether fields are present instead of using truthiness, so an age of zero is not silently omitted. It aliases attribute names and rejects an empty update. The existence condition avoids accidentally inserting an incomplete item during an update.

The list operation returns a continuation token. Follow it until null; one Scan cannot fetch an arbitrarily large table. For an email lookup, use Query on ByEmail and grant index access. A GSI does not enforce email uniqueness. Add application authorization, field-length checks and an appropriate idempotency design before using this API beyond a development account.

Step 4: connect Lambda and API Gateway

Add this after the table import in index.ts. It uses explicit provider resources so the IAM and invocation permission are visible:

import * as pulumi from "@pulumi/pulumi"
import * as aws from "@pulumi/aws"

const role = new aws.iam.Role("users-api-role", {
  assumeRolePolicy: JSON.stringify({
    Version: "2012-10-17",
    Statement: [{ Effect: "Allow", Principal: { Service: "lambda.amazonaws.com" }, Action: "sts:AssumeRole" }],
  }),
})
const logs = new aws.iam.RolePolicyAttachment("users-api-logs", {
  role: role.name,
  policyArn: "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
})
const access = new aws.iam.RolePolicy("users-api-data", {
  role: role.id,
  policy: Users.arn.apply((arn) => JSON.stringify({
    Version: "2012-10-17",
    Statement: [{
      Effect: "Allow",
      Action: ["dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Scan"],
      Resource: arn,
    }],
  })),
})
const handler = new aws.lambda.Function("users-api", {
  runtime: "nodejs24.x",
  handler: "index.handler",
  role: role.arn,
  code: new pulumi.asset.FileArchive("./lambda"),
  timeout: 10,
  environment: { variables: { TABLE_NAME: Users.name } },
}, { dependsOn: [logs, access] })

const api = new aws.apigatewayv2.Api("users-api", { protocolType: "HTTP" })
const integration = new aws.apigatewayv2.Integration("users-lambda", {
  apiId: api.id,
  integrationType: "AWS_PROXY",
  integrationUri: handler.invokeArn,
  payloadFormatVersion: "2.0",
})
for (const [name, routeKey] of Object.entries({
  create: "POST /users", list: "GET /users",
  update: "PATCH /users/{id}/{createdAt}", remove: "DELETE /users/{id}/{createdAt}",
})) {
  new aws.apigatewayv2.Route(name, {
    apiId: api.id,
    routeKey,
    target: pulumi.interpolate`integrations/${integration.id}`,
  })
}
new aws.apigatewayv2.Stage("default", { apiId: api.id, name: "$default", autoDeploy: true })
new aws.lambda.Permission("allow-api", {
  action: "lambda:InvokeFunction",
  function: handler.name,
  principal: "apigateway.amazonaws.com",
  sourceArn: pulumi.interpolate`${api.executionArn}/*/*`,
})
export const apiUrl = api.apiEndpoint
import * as pulumi from "@pulumi/pulumi"
import * as aws from "@pulumi/aws"

const role = new aws.iam.Role("users-api-role", {
  assumeRolePolicy: JSON.stringify({
    Version: "2012-10-17",
    Statement: [{ Effect: "Allow", Principal: { Service: "lambda.amazonaws.com" }, Action: "sts:AssumeRole" }],
  }),
})
const logs = new aws.iam.RolePolicyAttachment("users-api-logs", {
  role: role.name,
  policyArn: "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
})
const access = new aws.iam.RolePolicy("users-api-data", {
  role: role.id,
  policy: Users.arn.apply((arn) => JSON.stringify({
    Version: "2012-10-17",
    Statement: [{
      Effect: "Allow",
      Action: ["dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Scan"],
      Resource: arn,
    }],
  })),
})
const handler = new aws.lambda.Function("users-api", {
  runtime: "nodejs24.x",
  handler: "index.handler",
  role: role.arn,
  code: new pulumi.asset.FileArchive("./lambda"),
  timeout: 10,
  environment: { variables: { TABLE_NAME: Users.name } },
}, { dependsOn: [logs, access] })

const api = new aws.apigatewayv2.Api("users-api", { protocolType: "HTTP" })
const integration = new aws.apigatewayv2.Integration("users-lambda", {
  apiId: api.id,
  integrationType: "AWS_PROXY",
  integrationUri: handler.invokeArn,
  payloadFormatVersion: "2.0",
})
for (const [name, routeKey] of Object.entries({
  create: "POST /users", list: "GET /users",
  update: "PATCH /users/{id}/{createdAt}", remove: "DELETE /users/{id}/{createdAt}",
})) {
  new aws.apigatewayv2.Route(name, {
    apiId: api.id,
    routeKey,
    target: pulumi.interpolate`integrations/${integration.id}`,
  })
}
new aws.apigatewayv2.Stage("default", { apiId: api.id, name: "$default", autoDeploy: true })
new aws.lambda.Permission("allow-api", {
  action: "lambda:InvokeFunction",
  function: handler.name,
  principal: "apigateway.amazonaws.com",
  sourceArn: pulumi.interpolate`${api.executionArn}/*/*`,
})
export const apiUrl = api.apiEndpoint

The archive includes index.mjs, its package files and installed dependencies at the ZIP root. Inspect it before deployment and keep unrelated files out of lambda. Runtime code receives the resolved table name through an environment variable; it does not need Pulumi installed.

These routes have no authorizer. Keep the demonstration isolated, or add authorization before real data is written. The Lambda role restricts AWS resource access but does not decide which caller owns an item.

Step 5: process table changes

Streams can contain keys only, new images, old images, or both. The table above chooses both. Create stream/index.mjs:

export const handler = async (event) => {
  for (const record of event.Records) {
    console.log({ eventId: record.eventID, eventName: record.eventName, keys: record.dynamodb?.Keys })
  }
}
export const handler = async (event) => {
  for (const record of event.Records) {
    console.log({ eventId: record.eventID, eventName: record.eventName, keys: record.dynamodb?.Keys })
  }
}

Add a separate role, function and event source mapping to index.ts:

const streamRole = new aws.iam.Role("users-stream-role", {
  assumeRolePolicy: JSON.stringify({
    Version: "2012-10-17",
    Statement: [{ Effect: "Allow", Principal: { Service: "lambda.amazonaws.com" }, Action: "sts:AssumeRole" }],
  }),
})
const streamAccess = new aws.iam.RolePolicyAttachment("users-stream-access", {
  role: streamRole.name,
  policyArn: "arn:aws:iam::aws:policy/service-role/AWSLambdaDynamoDBExecutionRole",
})
const subscriber = new aws.lambda.Function("users-stream", {
  runtime: "nodejs24.x",
  handler: "index.handler",
  role: streamRole.arn,
  code: new pulumi.asset.FileArchive("./stream"),
}, { dependsOn: [streamAccess] })
new aws.lambda.EventSourceMapping("users-changes", {
  eventSourceArn: Users.streamArn,
  functionName: subscriber.arn,
  startingPosition: "TRIM_HORIZON",
  batchSize: 100,
  bisectBatchOnFunctionError: true,
  maximumRetryAttempts: 5,
}, { dependsOn: [streamAccess] })
const streamRole = new aws.iam.Role("users-stream-role", {
  assumeRolePolicy: JSON.stringify({
    Version: "2012-10-17",
    Statement: [{ Effect: "Allow", Principal: { Service: "lambda.amazonaws.com" }, Action: "sts:AssumeRole" }],
  }),
})
const streamAccess = new aws.iam.RolePolicyAttachment("users-stream-access", {
  role: streamRole.name,
  policyArn: "arn:aws:iam::aws:policy/service-role/AWSLambdaDynamoDBExecutionRole",
})
const subscriber = new aws.lambda.Function("users-stream", {
  runtime: "nodejs24.x",
  handler: "index.handler",
  role: streamRole.arn,
  code: new pulumi.asset.FileArchive("./stream"),
}, { dependsOn: [streamAccess] })
new aws.lambda.EventSourceMapping("users-changes", {
  eventSourceArn: Users.streamArn,
  functionName: subscriber.arn,
  startingPosition: "TRIM_HORIZON",
  batchSize: 100,
  bisectBatchOnFunctionError: true,
  maximumRetryAttempts: 5,
}, { dependsOn: [streamAccess] })

The managed role policy grants stream-read and logging permissions broadly. Replace it with a stream-scoped policy if that scope is too wide for your environment. Process every record in the batch, not only Records[0]. Add a failure destination and idempotent side effects before using the consumer for important processing; the Streams guide covers those settings.

Deploy, test and retain the data

pulumi preview --stack dev
pulumi up --stack dev
pulumi stack output apiUrl
pulumi preview --stack dev
pulumi up --stack dev
pulumi stack output apiUrl

Use the returned URL to POST a user, then keep both id and numeric createdAt from the response. PATCH and DELETE use both in the path. GET /users returns a page; follow its token for more. Check CloudWatch for stream events after writes. A correctly declared auto-deploying API stage should not need an unexplained second deployment.

These screenshots illustrate preview and deployment only; the resources and routes above replace the older callback-function implementation. For local SDK tests, use DynamoDB Local with an explicit endpoint and dummy credentials. Provider deployment, IAM and global replication still require AWS validation.

Pulumi protect rejects deletion through Pulumi, while retainOnDelete preserves the physical resource if it is removed from state through a permitted deletion. DynamoDB deletion protection is a third, service-side guard. Review the protect and retainOnDelete semantics before cleanup. Disable protections only as part of a reviewed retention or migration plan, and remember that retained tables continue to incur charges.