DynamoDB and Python: Boto3 Query and CRUD Cheat Sheet

Boto3 is the AWS SDK for Python. For DynamoDB, its resource interface reads and writes Python dictionaries through Table objects, while its client interface exposes the service APIs using DynamoDB's typed attribute values. Use Key for query key conditions, Attr for filters and conditions, and pagination whenever a Query or Scan can span multiple responses.

What is boto3, and should you use a resource or client?

Both interfaces are part of boto3. The resource is convenient for established item operations, including automatic batch writing. AWS has frozen new feature development for resources; existing resource APIs continue to work, while newer features are available through clients.

Do not pass already serialized attributes into a resource operation: it will treat those descriptors as ordinary dictionaries. Conversely, a separately created client requires typed values. The examples below use distinct resource and client objects to keep that boundary clear.

Cheat-sheet contents

• Connect to DynamoDB
• Create a table
• Delete a table
• List tables
• Get all items with Scan
• Get one item
• Batch get with retries
• Put an item
• Batch write items
• Query a set of items
• Query an index
• Update an item
• Conditionally update an item
• Increment an attribute
• Delete one item
• Delete all items
• Query with sorting
• Query and scan pagination
• Decimal and serialization
• Transactions
• Error handling
• DynamoDB Local

Connect boto3 to DynamoDB

Current boto3 requires Python 3.10 or later. Use a virtual environment and install boto3 there. For new Lambda examples, choose a supported runtime such as Python 3.13 and package the dependencies you need.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install boto3
python3 -m venv .venv
source .venv/bin/activate
python -m pip install boto3

Run this setup once before the later Python snippets. They share the Movies table, with Number partition key year and String sort key title. A GenreIndex with String partition key genre and String sort key title supports the index example.

import boto3
from boto3.dynamodb.conditions import Key, Attr
from botocore.config import Config
from decimal import Decimal

session = boto3.Session(region_name="us-east-1")
config = Config(retries={"mode": "standard", "total_max_attempts": 3})
dynamodb = session.resource("dynamodb", config=config)
client = session.client("dynamodb", config=config)
table = dynamodb.Table("Movies")
import boto3
from boto3.dynamodb.conditions import Key, Attr
from botocore.config import Config
from decimal import Decimal

session = boto3.Session(region_name="us-east-1")
config = Config(retries={"mode": "standard", "total_max_attempts": 3})
dynamodb = session.resource("dynamodb", config=config)
client = session.client("dynamodb", config=config)
table = dynamodb.Table("Movies")

Use your AWS profile during local development, for example by setting AWS_PROFILE, or let a Lambda execution role provide credentials. Do not embed access keys in application code. Constructing Table("Movies") creates a local resource object; it does not create the remote table.

The boto3 DynamoDB guide documents the resource API. Equivalent examples are available for Node.js and Java.

Create a table

The key schema and attribute definitions must agree on names and types. Define only table and index keys, not every attribute you intend to store. This creates an on-demand table and waits for it to become active.

table = dynamodb.create_table(
    TableName="Movies",
    BillingMode="PAY_PER_REQUEST",
    KeySchema=[
        {"AttributeName": "year", "KeyType": "HASH"},
        {"AttributeName": "title", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "year", "AttributeType": "N"},
        {"AttributeName": "title", "AttributeType": "S"},
        {"AttributeName": "genre", "AttributeType": "S"},
    ],
    GlobalSecondaryIndexes=[{
        "IndexName": "GenreIndex",
        "KeySchema": [
            {"AttributeName": "genre", "KeyType": "HASH"},
            {"AttributeName": "title", "KeyType": "RANGE"},
        ],
        "Projection": {"ProjectionType": "ALL"},
    }],
)
table.wait_until_exists()
print("Table is ready:", table.name)
table = dynamodb.create_table(
    TableName="Movies",
    BillingMode="PAY_PER_REQUEST",
    KeySchema=[
        {"AttributeName": "year", "KeyType": "HASH"},
        {"AttributeName": "title", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "year", "AttributeType": "N"},
        {"AttributeName": "title", "AttributeType": "S"},
        {"AttributeName": "genre", "AttributeType": "S"},
    ],
    GlobalSecondaryIndexes=[{
        "IndexName": "GenreIndex",
        "KeySchema": [
            {"AttributeName": "genre", "KeyType": "HASH"},
            {"AttributeName": "title", "KeyType": "RANGE"},
        ],
        "Projection": {"ProjectionType": "ALL"},
    }],
)
table.wait_until_exists()
print("Table is ready:", table.name)

A client-based equivalent waits with client.get_waiter("table_exists").wait(TableName="Movies"). Waiter timeouts are failures to finish waiting, not proof that the table will never become active.

Delete a table

Call delete() on a table resource, or delete_table() on a client. The service resource itself has no delete_table method. This removes all table data and indexes, so run it only when finished with the disposable example table.

client.delete_table(TableName="Movies")
client.get_waiter("table_not_exists").wait(TableName="Movies")
client.delete_table(TableName="Movies")
client.get_waiter("table_not_exists").wait(TableName="Movies")

Deletion protection blocks this request when enabled. Item deletion, covered below, leaves the table and its configuration intact.

List tables

A client paginator retrieves all table names for the configured account and Region. A single list_tables() response may be incomplete.

for page in client.get_paginator("list_tables").paginate():
    for name in page.get("TableNames", []):
        print(name)
for page in client.get_paginator("list_tables").paginate():
    for name in page.get("TableNames", []):
        print(name)

The resource alternative, dynamodb.tables.all(), also handles pagination when iterated. Avoid turning it into a list unless you need all resource objects in memory.

Get all items with Scan

A Scan reads the table page by page. The helper below also accepts query, so later examples can reuse it without losing conditions, index names or filters on subsequent requests.

def read_pages(operation, **params):
    while True:
        response = operation(**params)
        yield response
        last_key = response.get("LastEvaluatedKey")
        if not last_key:
            break
        params["ExclusiveStartKey"] = last_key

for page in read_pages(table.scan):
    for item in page.get("Items", []):
        print(item)
def read_pages(operation, **params):
    while True:
        response = operation(**params)
        yield response
        last_key = response.get("LastEvaluatedKey")
        if not last_key:
            break
        params["ExclusiveStartKey"] = last_key

for page in read_pages(table.scan):
    for item in page.get("Items", []):
        print(item)

Filter with Attr for non-key attributes. Keep the filter on every page, including pages after an empty result:

for page in read_pages(
    table.scan,
    FilterExpression=Attr("country").eq("US") & Attr("city").eq("NYC"),
):
    for item in page.get("Items", []):
        print(item)
for page in read_pages(
    table.scan,
    FilterExpression=Attr("country").eq("US") & Attr("city").eq("NYC"),
):
    for item in page.get("Items", []):
        print(item)

DynamoDB evaluates up to 1 MB before filtering. Filters reduce returned data, not the read capacity spent evaluating it. Even a strongly consistent scan does not provide a table-wide snapshot. Use Query when the access pattern identifies a partition.

Get one item

Supply the full primary key with its correct types. The resource returns DynamoDB Numbers as Decimal, and an absent item is represented by a response without Item.

response = table.get_item(
    Key={"year": 2026, "title": "Example Movie"},
    ConsistentRead=True,
)
print(response.get("Item"))
response = table.get_item(
    Key={"year": 2026, "title": "Example Movie"},
    ConsistentRead=True,
)
print(response.get("Item"))

get_item does not query a secondary index. Use Query for index lookups, then read the base item when you need attributes not projected into that index.

Batch get items with retries

BatchGet accepts up to 100 keys and returns at most 16 MB. It may return partial results with UnprocessedKeys even when the request succeeds. Retry only those keys with backoff; results are unordered, and missing items are omitted.

import random
import time

pending = {"Movies": {"Keys": [
    {"year": 2026, "title": "Example Movie"},
    {"year": 2026, "title": "Another Movie"},
], "ConsistentRead": True}}

for attempt in range(8):
    if attempt:
        time.sleep(random.uniform(0, min(5.0, 0.1 * 2 ** attempt)))
    response = dynamodb.batch_get_item(RequestItems=pending)
    for item in response.get("Responses", {}).get("Movies", []):
        print(item)
    pending = response.get("UnprocessedKeys", {})
    if not pending:
        break
else:
    # Retain pending in the calling job for a later retry.
    raise RuntimeError("Batch read incomplete", pending)
import random
import time

pending = {"Movies": {"Keys": [
    {"year": 2026, "title": "Example Movie"},
    {"year": 2026, "title": "Another Movie"},
], "ConsistentRead": True}}

for attempt in range(8):
    if attempt:
        time.sleep(random.uniform(0, min(5.0, 0.1 * 2 ** attempt)))
    response = dynamodb.batch_get_item(RequestItems=pending)
    for item in response.get("Responses", {}).get("Movies", []):
        print(item)
    pending = response.get("UnprocessedKeys", {})
    if not pending:
        break
else:
    # Retain pending in the calling job for a later retry.
    raise RuntimeError("Batch read incomplete", pending)

Deduplicate input keys and split larger inputs into separate requests. Automatic SDK retries for failed requests do not replace this loop for successful responses containing unfinished work.

Put an item

put_item replaces the entire item at the same key. Add a condition to make it create-only. Attributes that are not keys can vary between items.

table.put_item(
    Item={
        "year": 2026,
        "title": "Example Movie",
        "genre": "Drama",
        "rating": Decimal("8.5"),
        "country": "US",
        "city": "NYC",
    },
    ConditionExpression=Attr("year").not_exists(),
)
table.put_item(
    Item={
        "year": 2026,
        "title": "Example Movie",
        "genre": "Drama",
        "rating": Decimal("8.5"),
        "country": "US",
        "city": "NYC",
    },
    ConditionExpression=Attr("year").not_exists(),
)

The condition is checked against the item identified by both key attributes. An existing movie with another title in the same year does not block the insert.

Batch write items

batch_writer() buffers puts and deletes, splits them into batches, and resends unprocessed items. Exiting the context flushes remaining work. It does not provide transactions, conditions or partial attribute updates.

with table.batch_writer(overwrite_by_pkeys=["year", "title"]) as batch:
    batch.put_item(Item={
        "year": 2026, "title": "Another Movie", "genre": "Drama",
    })
    batch.put_item(Item={
        "year": 2025, "title": "Older Movie", "genre": "Comedy",
    })
    batch.delete_item(Key={"year": 2024, "title": "Retired Movie"})
with table.batch_writer(overwrite_by_pkeys=["year", "title"]) as batch:
    batch.put_item(Item={
        "year": 2026, "title": "Another Movie", "genre": "Drama",
    })
    batch.put_item(Item={
        "year": 2025, "title": "Older Movie", "genre": "Comedy",
    })
    batch.delete_item(Key={"year": 2024, "title": "Retired Movie"})

overwrite_by_pkeys deduplicates matching keys in the local buffer; it is not a server-side conditional write. Each put still replaces the full stored item. For jobs needing a strict deadline, checkpointing or custom backoff, implement a bounded client batch_write_item loop and retain UnprocessedItems. The batch writer reference describes the helper's behavior.

Query a set of items

Use Key for the partition-key equality and an optional sort-key predicate. Query can read a base table or an index; it does not require a secondary index. Use the read_pages helper from the Scan section to retrieve every page.

for page in read_pages(
    table.query,
    KeyConditionExpression=Key("year").eq(2026) & Key("title").begins_with("Example"),
    FilterExpression=Attr("rating").gte(Decimal("8.0")),
):
    for item in page.get("Items", []):
        print(item["title"], item.get("rating"))
for page in read_pages(
    table.query,
    KeyConditionExpression=Key("year").eq(2026) & Key("title").begins_with("Example"),
    FilterExpression=Attr("rating").gte(Decimal("8.0")),
):
    for item in page.get("Items", []):
        print(item["title"], item.get("rating"))

The key condition narrows what DynamoDB reads. The filter runs afterward and must not replace the required partition-key condition. See query examples for other access patterns.

Query an index

This uses the GenreIndex defined above. Its partition key is genre, so the key condition must use genre, not the base table's year.

for page in read_pages(
    table.query,
    IndexName="GenreIndex",
    KeyConditionExpression=Key("genre").eq("Drama"),
):
    for item in page.get("Items", []):
        print(item["title"])
for page in read_pages(
    table.query,
    IndexName="GenreIndex",
    KeyConditionExpression=Key("genre").eq("Drama"),
):
    for item in page.get("Items", []):
        print(item["title"])

Global secondary indexes allow only eventually consistent reads. Setting ConsistentRead=True on a GSI query fails. The example index projects all attributes; narrower projections return fewer attributes.

Update selected attributes

Use an expression to change attributes without replacing the whole item. Alias attribute names to avoid reserved-word conflicts. The existence condition prevents UpdateItem from creating an item when the key is absent.

response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="SET #country = :country",
    ExpressionAttributeNames={"#country": "country"},
    ExpressionAttributeValues={":country": "Canada"},
    ConditionExpression=Attr("year").exists(),
    ReturnValues="UPDATED_NEW",
)
print(response["Attributes"])
response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="SET #country = :country",
    ExpressionAttributeNames={"#country": "country"},
    ExpressionAttributeValues={":country": "Canada"},
    ConditionExpression=Attr("year").exists(),
    ReturnValues="UPDATED_NEW",
)
print(response["Attributes"])

You cannot change primary-key attributes with UpdateItem. Moving an item to another key requires writing a new item and deleting the old one, optionally in a transaction.

Conditionally update an item

A condition and update execute atomically. This changes the rating only if it still equals the expected value, so a concurrent writer cannot silently overwrite another change.

response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="SET #rating = :next",
    ExpressionAttributeNames={"#rating": "rating"},
    ExpressionAttributeValues={":next": Decimal("8.6")},
    ConditionExpression=Attr("rating").eq(Decimal("8.5")),
    ReturnValues="ALL_NEW",
)
print(response["Attributes"])
response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="SET #rating = :next",
    ExpressionAttributeNames={"#rating": "rating"},
    ExpressionAttributeValues={":next": Decimal("8.6")},
    ConditionExpression=Attr("rating").eq(Decimal("8.5")),
    ReturnValues="ALL_NEW",
)
print(response["Attributes"])

A false condition raises ClientError with code ConditionalCheckFailedException. Treat that as an application conflict, not a transient error to retry indefinitely.

Increment an attribute atomically

ADD increments a top-level Number without a read-modify-write race. A missing numeric attribute starts at zero. The condition below requires the movie itself to exist.

response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="ADD #views :one",
    ExpressionAttributeNames={"#views": "views"},
    ExpressionAttributeValues={":one": 1},
    ConditionExpression=Attr("year").exists(),
    ReturnValues="UPDATED_NEW",
)
print(response["Attributes"]["views"])
response = table.update_item(
    Key={"year": 2026, "title": "Example Movie"},
    UpdateExpression="ADD #views :one",
    ExpressionAttributeNames={"#views": "views"},
    ExpressionAttributeValues={":one": 1},
    ConditionExpression=Attr("year").exists(),
    ReturnValues="UPDATED_NEW",
)
print(response["Attributes"]["views"])

An atomic increment is not idempotent: retrying after an ambiguous timeout can increment twice. For nested counters, use SET #stats.#views = if_not_exists(#stats.#views, :zero) + :one with name aliases and values, and ensure the parent stats map already exists.

Delete one item

Delete with the full primary key. Without a condition, a missing item is a successful no-op. ALL_OLD returns the removed item when present.

response = table.delete_item(
    Key={"year": 2026, "title": "Example Movie"},
    ReturnValues="ALL_OLD",
)
print(response.get("Attributes"))
response = table.delete_item(
    Key={"year": 2026, "title": "Example Movie"},
    ReturnValues="ALL_OLD",
)
print(response.get("Attributes"))

Use a condition when deletion must depend on a rating, status or version that another writer could change.

Delete all items without deleting the table

Scan both key attributes and delete through the batch writer. This is destructive and is not an atomic truncate. Stop concurrent writers and take an appropriate backup before clearing a table; concurrent inserts can survive the scan, and concurrent replacements can be deleted.

for page in read_pages(
    table.scan,
    ProjectionExpression="#year, #title",
    ExpressionAttributeNames={"#year": "year", "#title": "title"},
):
    with table.batch_writer() as batch:
        for item in page.get("Items", []):
            batch.delete_item(Key={"year": item["year"], "title": item["title"]})
for page in read_pages(
    table.scan,
    ProjectionExpression="#year, #title",
    ExpressionAttributeNames={"#year": "year", "#title": "title"},
):
    with table.batch_writer() as batch:
        for item in page.get("Items", []):
            batch.delete_item(Key={"year": item["year"], "title": item["title"]})

Projection reduces response size, not read capacity. Keep writers stopped while verifying the table is empty. The delete-all-items guide covers alternatives and operational trade-offs.

Query with sorting

Query orders results by the sort key within the selected partition. ScanIndexForward=False reverses the order. It does not sort by an arbitrary attribute such as rating, and Scan has no equivalent ordering option.

for page in read_pages(
    table.query,
    KeyConditionExpression=Key("year").eq(2026),
    ScanIndexForward=False,
):
    for item in page.get("Items", []):
        print(item["title"])
for page in read_pages(
    table.query,
    KeyConditionExpression=Key("year").eq(2026),
    ScanIndexForward=False,
):
    for item in page.get("Items", []):
        print(item["title"])

Here the sort key is a String, so titles follow UTF-8 byte order. To sort by another attribute efficiently, design an index with that attribute as its sort key.

Query and scan with client paginators

Resource loops use LastEvaluatedKey as the next request's ExclusiveStartKey, as read_pages does. Clients also offer paginators. The separately created client below uses typed values and returns typed items, so deserialize each attribute.

from boto3.dynamodb.types import TypeDeserializer

deserializer = TypeDeserializer()
for page in client.get_paginator("query").paginate(
    TableName="Movies",
    KeyConditionExpression="#year = :year",
    ExpressionAttributeNames={"#year": "year"},
    ExpressionAttributeValues={":year": {"N": "2026"}},
    PaginationConfig={"PageSize": 100},
):
    for raw_item in page.get("Items", []):
        item = {name: deserializer.deserialize(value) for name, value in raw_item.items()}
        print(item)

for page in client.get_paginator("scan").paginate(TableName="Movies"):
    for raw_item in page.get("Items", []):
        print({name: deserializer.deserialize(value) for name, value in raw_item.items()})
from boto3.dynamodb.types import TypeDeserializer

deserializer = TypeDeserializer()
for page in client.get_paginator("query").paginate(
    TableName="Movies",
    KeyConditionExpression="#year = :year",
    ExpressionAttributeNames={"#year": "year"},
    ExpressionAttributeValues={":year": {"N": "2026"}},
    PaginationConfig={"PageSize": 100},
):
    for raw_item in page.get("Items", []):
        item = {name: deserializer.deserialize(value) for name, value in raw_item.items()}
        print(item)

for page in client.get_paginator("scan").paginate(TableName="Movies"):
    for raw_item in page.get("Items", []):
        print({name: deserializer.deserialize(value) for name, value in raw_item.items()})

PageSize controls the number of evaluated items per request, not the number of filtered matches. MaxItems limits the paginator's total results and can stop early. Follow continuation state even when a page contains no matches. See the Query paginator reference.

Decimal values and DynamoDB JSON

The resource serializer accepts integers and Decimal, but rejects Python float. Construct decimals from strings to avoid carrying binary floating-point errors into the value. Parse JSON numbers directly as Decimal when importing data.

import json
from boto3.dynamodb.types import TypeSerializer, TypeDeserializer

item = json.loads(
    '{"year": 2026, "title": "Imported Movie", "rating": 8.5}',
    parse_float=Decimal,
)
serializer = TypeSerializer()
deserializer = TypeDeserializer()
wire_item = {name: serializer.serialize(value) for name, value in item.items()}
client.put_item(TableName="Movies", Item=wire_item)
plain_item = {name: deserializer.deserialize(value) for name, value in wire_item.items()}

def decimal_json(value):
    if isinstance(value, Decimal):
        return str(value)
    raise TypeError(f"Unsupported JSON value: {type(value).__name__}")

print(json.dumps(plain_item, default=decimal_json))
import json
from boto3.dynamodb.types import TypeSerializer, TypeDeserializer

item = json.loads(
    '{"year": 2026, "title": "Imported Movie", "rating": 8.5}',
    parse_float=Decimal,
)
serializer = TypeSerializer()
deserializer = TypeDeserializer()
wire_item = {name: serializer.serialize(value) for name, value in item.items()}
client.put_item(TableName="Movies", Item=wire_item)
plain_item = {name: deserializer.deserialize(value) for name, value in wire_item.items()}

def decimal_json(value):
    if isinstance(value, Decimal):
        return str(value)
    raise TypeError(f"Unsupported JSON value: {type(value).__name__}")

print(json.dumps(plain_item, default=decimal_json))

This JSON encoder deliberately emits decimals as strings to preserve precision. Choose a different explicit representation if your API contract requires numeric JSON; converting everything to float can lose information. Sets and binary values also need their own JSON representation. The boto3 type conversion source lists supported conversions.

Write and read transactions through the client

A write transaction applies all operations or none. Use the client with typed attributes; the resource interface has no transaction method. This example creates two different movie items with create-only conditions.

from boto3.dynamodb.types import TypeSerializer

serializer = TypeSerializer()
def serialize_item(item):
    return {name: serializer.serialize(value) for name, value in item.items()}

request = {
    "ClientRequestToken": "movie-pair-2026-v1",
    "TransactItems": [
        {"Put": {
            "TableName": "Movies",
            "Item": serialize_item({"year": 2026, "title": title, "genre": "Drama"}),
            "ConditionExpression": "attribute_not_exists(#year)",
            "ExpressionAttributeNames": {"#year": "year"},
        }}
        for title in ["Transaction Movie A", "Transaction Movie B"]
    ],
}
client.transact_write_items(**request)
from boto3.dynamodb.types import TypeSerializer

serializer = TypeSerializer()
def serialize_item(item):
    return {name: serializer.serialize(value) for name, value in item.items()}

request = {
    "ClientRequestToken": "movie-pair-2026-v1",
    "TransactItems": [
        {"Put": {
            "TableName": "Movies",
            "Item": serialize_item({"year": 2026, "title": title, "genre": "Drama"}),
            "ConditionExpression": "attribute_not_exists(#year)",
            "ExpressionAttributeNames": {"#year": "year"},
        }}
        for title in ["Transaction Movie A", "Transaction Movie B"]
    ],
}
client.transact_write_items(**request)

Use a distinct token for each logical write, retaining that token and identical parameters across retries. Its idempotency window is 10 minutes. Transactions allow up to 100 unique items and 4 MB within one account and Region. You cannot target the same item twice, and MRSC global tables do not support transactions. See the transaction API.

Read the pair atomically using transact_get_items; results follow the requested order and contain typed attributes:

from boto3.dynamodb.types import TypeDeserializer

deserializer = TypeDeserializer()
response = client.transact_get_items(TransactItems=[
    {"Get": {
        "TableName": "Movies",
        "Key": {"year": {"N": "2026"}, "title": {"S": title}},
    }}
    for title in ["Transaction Movie A", "Transaction Movie B"]
])
for result in response["Responses"]:
    raw_item = result.get("Item")
    print(None if raw_item is None else {
        name: deserializer.deserialize(value) for name, value in raw_item.items()
    })
from boto3.dynamodb.types import TypeDeserializer

deserializer = TypeDeserializer()
response = client.transact_get_items(TransactItems=[
    {"Get": {
        "TableName": "Movies",
        "Key": {"year": {"N": "2026"}, "title": {"S": title}},
    }}
    for title in ["Transaction Movie A", "Transaction Movie B"]
])
for result in response["Responses"]:
    raw_item = result.get("Item")
    print(None if raw_item is None else {
        name: deserializer.deserialize(value) for name, value in raw_item.items()
    })

Handle errors by ClientError code

Catch ClientError for service errors and inspect response["Error"]["Code"]. Do not parse message text or silently turn every exception into an empty result.

from botocore.exceptions import ClientError

try:
    table.put_item(
        Item={"year": 2026, "title": "Example Movie"},
        ConditionExpression=Attr("year").not_exists(),
    )
except ClientError as error:
    code = error.response["Error"]["Code"]
    if code == "ConditionalCheckFailedException":
        print("Movie already exists")
    else:
        print(code, error.response.get("ResponseMetadata", {}).get("RequestId"))
        raise
from botocore.exceptions import ClientError

try:
    table.put_item(
        Item={"year": 2026, "title": "Example Movie"},
        ConditionExpression=Attr("year").not_exists(),
    )
except ClientError as error:
    code = error.response["Error"]["Code"]
    if code == "ConditionalCheckFailedException":
        print("Movie already exists")
    else:
        print(code, error.response.get("ResponseMetadata", {}).get("RequestId"))
        raise

TransactionCanceledException requires examining the transaction's failure, not treating it as a successful write. Credential errors and local parameter validation errors are botocore exceptions outside ClientError. SDK retries handle some transient failures; they do not fix invalid keys or failed conditions. The error handling reference explains the distinction.

Use boto3 with DynamoDB Local

Start DynamoDB Local with Docker, then override the endpoint on both interfaces. These dummy credentials are only for the local endpoint.

docker run --rm -p 127.0.0.1:8000:8000 amazon/dynamodb-local:3.3.2 \
  -jar DynamoDBLocal.jar -inMemory -sharedDb
docker run --rm -p 127.0.0.1:8000:8000 amazon/dynamodb-local:3.3.2 \
  -jar DynamoDBLocal.jar -inMemory -sharedDb

local_session = boto3.Session(
    region_name="us-east-1",
    aws_access_key_id="local",
    aws_secret_access_key="local",
)
dynamodb = local_session.resource("dynamodb", endpoint_url="http://localhost:8000")
client = local_session.client("dynamodb", endpoint_url="http://localhost:8000")
table = dynamodb.Table("Movies")
local_session = boto3.Session(
    region_name="us-east-1",
    aws_access_key_id="local",
    aws_secret_access_key="local",
)
dynamodb = local_session.resource("dynamodb", endpoint_url="http://localhost:8000")
client = local_session.client("dynamodb", endpoint_url="http://localhost:8000")
table = dynamodb.Table("Movies")

Create the table against this endpoint before running item operations. This in-memory container loses data when stopped and cannot reproduce every cloud behavior. See running DynamoDB locally for persistent storage and environment setup.

Frequently asked questions