Amazon DynamoDB
Everyday operations for reading, writing, and querying DynamoDB tables across Python and TypeScript.
//
Search across all documentation pages
Everyday operations for reading, writing, and querying DynamoDB tables across Python and TypeScript.
Set up a DynamoDB table with a partition key and optional sort key.
# --- Python (boto3) ---
import boto3
ddb = boto3.client("dynamodb")
ddb.create_table(
TableName="Users",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST"
)// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, CreateTableCommand } from "@aws-sdk/client-dynamodb";
const client = new DynamoDBClient({});
await client.send(new CreateTableCommand({
TableName: "Users",
KeySchema: [{ AttributeName: "id", KeyType: "HASH" }],
AttributeDefinitions: [{ AttributeName: "id", AttributeType: "S" }],
BillingMode: "PAY_PER_REQUEST"
}));Insert or overwrite an item in a table.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
ddb.put_item(Item={"id": "u1", "name": "Ada", "email": "ada@example.com"})// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
await doc.send(new PutCommand({
TableName: "Users",
Item: { id: "u1", name: "Ada", email: "ada@example.com" }
}));Fetch a single item by its primary key.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
response = ddb.get_item(Key={"id": "u1"})
item = response.get("Item") # {'id': 'u1', 'name': 'Ada', ...}// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, GetCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const { Item } = await doc.send(new GetCommand({
TableName: "Users",
Key: { id: "u1" }
}));Fetch all items with a specific partition key value.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
response = ddb.query(KeyConditionExpression="id = :id", ExpressionAttributeValues={":id": "u1"})
items = response.get("Items", [])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, QueryCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const { Items } = await doc.send(new QueryCommand({
TableName: "Users",
KeyConditionExpression: "id = :id",
ExpressionAttributeValues: { ":id": "u1" }
}));Query using a Global Secondary Index instead of the primary key.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
response = ddb.query(
IndexName="EmailIndex",
KeyConditionExpression="email = :email",
ExpressionAttributeValues={":email": "ada@example.com"}
)
items = response.get("Items", [])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, QueryCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const { Items } = await doc.send(new QueryCommand({
TableName: "Users",
IndexName: "EmailIndex",
KeyConditionExpression: "email = :email",
ExpressionAttributeValues: { ":email": "ada@example.com" }
}));Retrieve items using a scan operation with optional filtering.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
response = ddb.scan(
FilterExpression="attribute_exists(#st) AND #st = :status",
ExpressionAttributeNames={"#st": "status"},
ExpressionAttributeValues={":status": "active"}
)
items = response.get("Items", [])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, ScanCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const { Items } = await doc.send(new ScanCommand({
TableName: "Users",
FilterExpression: "attribute_exists(#st) AND #st = :status",
ExpressionAttributeNames: { "#st": "status" },
ExpressionAttributeValues: { ":status": "active" }
}));Update specific attributes using SET (overwrite) or ADD (increment/append).
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
ddb.update_item(
Key={"id": "u1"},
UpdateExpression="SET #nm = :name, #cnt = if_not_exists(#cnt, :zero) + :inc",
ExpressionAttributeNames={"#nm": "name", "#cnt": "login_count"},
ExpressionAttributeValues={":name": "Ada Lovelace", ":zero": 0, ":inc": 1}
)// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, UpdateCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
await doc.send(new UpdateCommand({
TableName: "Users",
Key: { id: "u1" },
UpdateExpression: "SET #nm = :name, #cnt = if_not_exists(#cnt, :zero) + :inc",
ExpressionAttributeNames: { "#nm": "name", "#cnt": "login_count" },
ExpressionAttributeValues: { ":name": "Ada Lovelace", ":zero": 0, ":inc": 1 }
}));Write only if a condition is true (e.g., attribute does not exist).
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
ddb.put_item(
Item={"id": "u2", "name": "Grace"},
ConditionExpression="attribute_not_exists(id)"
)// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
await doc.send(new PutCommand({
TableName: "Users",
Item: { id: "u2", name: "Grace" },
ConditionExpression: "attribute_not_exists(id)"
}));Remove a single item from the table.
# --- Python (boto3) ---
import boto3
ddb = boto3.resource("dynamodb").Table("Users")
ddb.delete_item(Key={"id": "u1"})// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, DeleteCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
await doc.send(new DeleteCommand({
TableName: "Users",
Key: { id: "u1" }
}));Write or delete multiple items in a single request.
# --- Python (boto3) ---
import boto3
ddb = boto3.client("dynamodb")
ddb.batch_write_item(RequestItems={
"Users": [
{"PutRequest": {"Item": {"id": {"S": "u1"}, "name": {"S": "Ada"}}}},
{"PutRequest": {"Item": {"id": {"S": "u2"}, "name": {"S": "Grace"}}}}
]
})// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, BatchWriteCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
await doc.send(new BatchWriteCommand({
RequestItems: {
Users: [
{ PutRequest: { Item: { id: "u1", name: "Ada" } } },
{ PutRequest: { Item: { id: "u2", name: "Grace" } } }
]
}
}));Retrieve multiple items in a single request.
# --- Python (boto3) ---
import boto3
ddb = boto3.client("dynamodb")
response = ddb.batch_get_item(
RequestItems={"Users": {"Keys": [{"id": {"S": "u1"}}, {"id": {"S": "u2"}}]}}
)
items = response["Responses"].get("Users", [])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, BatchGetCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const { Responses } = await doc.send(new BatchGetCommand({
RequestItems: {
Users: { Keys: [{ id: "u1" }, { id: "u2" }] }
}
}));Perform multiple writes atomically across items and tables.
# --- Python (boto3) ---
import boto3
ddb = boto3.client("dynamodb")
ddb.transact_write_items(TransactItems=[
{"Put": {"TableName": "Users", "Item": {"id": {"S": "u1"}, "name": {"S": "Ada"}}}},
{"Put": {"TableName": "Users", "Item": {"id": {"S": "u2"}, "name": {"S": "Grace"}}}}
])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, TransactWriteItemsCommand } from "@aws-sdk/client-dynamodb";
const client = new DynamoDBClient({});
await client.send(new TransactWriteItemsCommand({
TransactItems: [
{ Put: { TableName: "Users", Item: { id: { S: "u1" }, name: { S: "Ada" } } } },
{ Put: { TableName: "Users", Item: { id: { S: "u2" }, name: { S: "Grace" } } } }
]
}));Compare native AttributeValue format with the higher-level document client for cleaner code.
# --- Python (boto3) ---
# Native: requires AttributeValue wrapping
ddb_low = boto3.client("dynamodb")
ddb_low.put_item(TableName="Users", Item={"id": {"S": "u1"}, "age": {"N": "30"}})
# Document client: clean native Python types
ddb_high = boto3.resource("dynamodb").Table("Users")
ddb_high.put_item(Item={"id": "u1", "age": 30})// --- TypeScript (AWS SDK v3) ---
// Native: requires AttributeValue wrapping
import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
const client = new DynamoDBClient({});
await client.send(new PutItemCommand({
TableName: "Users",
Item: { id: { S: "u1" }, age: { N: "30" } }
}));
// Document client: clean native JS types
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
const doc = DynamoDBDocumentClient.from(client);
await doc.send(new PutCommand({
TableName: "Users",
Item: { id: "u1", age: 30 }
}));Handle pagination when query results exceed the DynamoDB page size limit.
# --- Python (boto3) ---
import boto3
ddb = boto3.client("dynamodb")
paginator = ddb.get_paginator("query")
pages = paginator.paginate(
TableName="Users",
KeyConditionExpression="id = :id",
ExpressionAttributeValues={":id": {"S": "u1"}}
)
for page in pages:
items = page.get("Items", [])// --- TypeScript (AWS SDK v3) ---
import { DynamoDBDocumentClient, QueryCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
const doc = DynamoDBDocumentClient.from(new DynamoDBClient({}));
let lastKey: any;
do {
const { Items, LastEvaluatedKey } = await doc.send(new QueryCommand({
TableName: "Users",
KeyConditionExpression: "id = :id",
ExpressionAttributeValues: { ":id": "u1" },
ExclusiveStartKey: lastKey
}));
lastKey = LastEvaluatedKey;
} while (lastKey);Process record changes from a DynamoDB Stream via Lambda or consumer.
# --- Python (boto3) ---
# Lambda handler receiving stream records
def handler(event, context):
for record in event["Records"]:
if record["eventName"] == "MODIFY":
new_image = record["dynamodb"]["NewImage"]
old_image = record["dynamodb"]["OldImage"]
elif record["eventName"] == "REMOVE":
key = record["dynamodb"]["Keys"]// --- TypeScript (AWS SDK v3) ---
// Lambda handler receiving stream records
export const handler = async (event: any) => {
for (const record of event.Records) {
if (record.eventName === "MODIFY") {
const newImage = record.dynamodb.NewImage;
const oldImage = record.dynamodb.OldImage;
} else if (record.eventName === "REMOVE") {
const key = record.dynamodb.Keys;
}
}
};Stack versions: This page was written for boto3 1.43.x (Python 3.10+) and the AWS SDK for JavaScript v3 (Node.js 18+).
Reviewed by Chris St. John·Last updated Jul 23, 2026