DynamoDB: Items, Expressions, GSI and Transactions Reference Guide

# AWS CLI (already-installed env)
aws --version
aws configure                # access key / secret / region

# Local-only dev — no AWS account needed
docker run -d --name ddb \
    -p 8000:8000 \
    amazon/dynamodb-local:latest \
    -jar DynamoDBLocal.jar -sharedDb

# Point the CLI at the local endpoint
aws dynamodb list-tables --endpoint-url http://localhost:8000

# Python SDK
pip install "boto3>=1.34"

python - <<'PY'
import boto3
ddb = boto3.resource("dynamodb", endpoint_url="http://localhost:8000",
                    region_name="us-east-1")
print([t.name for t in ddb.tables.all()])
PY

# Browse / model visually with NoSQL Workbench (separate desktop tool)
# https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/workbench.html

# Orders table — composite key (customer_id + order_id) and a GSI on status.
# Single-table design: PK is the entity identifier, SK is the entity sort.
aws dynamodb create-table \
    --table-name orders \
    --attribute-definitions \
        AttributeName=pk,AttributeType=S \
        AttributeName=sk,AttributeType=S \
        AttributeName=status,AttributeType=S \
        AttributeName=created,AttributeType=S \
    --key-schema \
        AttributeName=pk,KeyType=HASH \
        AttributeName=sk,KeyType=RANGE \
    --billing-mode PAY_PER_REQUEST \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
    --global-secondary-indexes '[
        {
          "IndexName":   "gsi_status",
          "KeySchema":   [
            {"AttributeName":"status","KeyType":"HASH"},
            {"AttributeName":"created","KeyType":"RANGE"}
          ],
          "Projection":  {"ProjectionType":"ALL"}
        }
    ]'

# Wait until ACTIVE (and the GSI is ACTIVE too)
aws dynamodb wait table-exists --table-name orders

# Enable TTL on a numeric "expires" attribute (epoch seconds)
aws dynamodb update-time-to-live --table-name orders \
    --time-to-live-specification 'Enabled=true,AttributeName=expires'

import boto3
from boto3.dynamodb.conditions import Key, Attr
from decimal import Decimal

ddb   = boto3.resource("dynamodb", region_name="us-east-1")
table = ddb.Table("orders")

# Create (only if order_id is new) — atomic check-and-insert
table.put_item(
    Item={
        "pk":       "CUSTOMER#42",
        "sk":       "ORDER#2026-05-19-abc",
        "status":   "pending",
        "created":  "2026-05-19T10:00:00Z",
        "total":    Decimal("19.95"),    # never float — precision matters for money
        "tags":     {"web", "first-order"},
    },
    ConditionExpression="attribute_not_exists(sk)"
)

# Read — strongly consistent
order = table.get_item(
    Key={"pk": "CUSTOMER#42", "sk": "ORDER#2026-05-19-abc"},
    ConsistentRead=True,
)["Item"]

# Update — atomic counter + status change, return the new state
table.update_item(
    Key={"pk": "CUSTOMER#42", "sk": "ORDER#2026-05-19-abc"},
    UpdateExpression="SET #s = :s, attempts = if_not_exists(attempts, :z) + :one",
    ConditionExpression="#s = :pending",
    ExpressionAttributeNames={"#s": "status"},
    ExpressionAttributeValues={":s":"paid", ":pending":"pending",
                               ":z": Decimal(0), ":one": Decimal(1)},
    ReturnValues="ALL_NEW",
)

# Batch writes — auto-retries unprocessed items
with table.batch_writer() as batch:
    for row in rows:
        batch.put_item(Item=row)

import boto3
from boto3.dynamodb.conditions import Key, Attr

ddb   = boto3.resource("dynamodb", region_name="us-east-1")
table = ddb.Table("orders")

# 1. All orders for one customer, newest first
resp = table.query(
    KeyConditionExpression = Key("pk").eq("CUSTOMER#42") & Key("sk").begins_with("ORDER#"),
    ScanIndexForward       = False,            # DESC by sort key
    Limit                  = 25,
    ProjectionExpression   = "sk, #s, total",
    ExpressionAttributeNames = {"#s": "status"},
)
items, next_key = resp["Items"], resp.get("LastEvaluatedKey")

# 2. Paginate the rest — paginator handles LastEvaluatedKey for you
paginator = table.meta.client.get_paginator("query")
for page in paginator.paginate(
    TableName              = "orders",
    KeyConditionExpression = "pk = :p AND begins_with(sk, :prefix)",
    ExpressionAttributeValues = {":p": {"S":"CUSTOMER#42"}, ":prefix": {"S":"ORDER#"}},
    PaginationConfig       = {"PageSize": 100},
):
    for item in page["Items"]:
        handle(item)

# 3. Query a GSI — all "paid" orders across customers in the last 7 days
resp = table.query(
    IndexName              = "gsi_status",
    KeyConditionExpression = Key("status").eq("paid") &
                             Key("created").gte("2026-05-12T00:00:00Z"),
    FilterExpression       = Attr("total").gte(50),
)

import boto3, uuid
from botocore.exceptions import ClientError

client = boto3.client("dynamodb", region_name="us-east-1")

# Atomic order placement:
#   1. Create the order row (only if it doesn't already exist)
#   2. Decrement stock on the product (only if stock >= qty)
# Either both succeed or neither does.
try:
    client.transact_write_items(
        TransactItems = [
            {"Put": {
                "TableName": "orders",
                "Item": {
                    "pk":     {"S": "CUSTOMER#42"},
                    "sk":     {"S": "ORDER#2026-05-19-abc"},
                    "status": {"S": "pending"},
                    "total":  {"N": "19.95"},
                },
                "ConditionExpression": "attribute_not_exists(sk)"
            }},
            {"Update": {
                "TableName": "inventory",
                "Key":        {"sku": {"S":"SKU-1"}},
                "UpdateExpression":   "SET stock = stock - :q",
                "ConditionExpression": "stock >= :q",
                "ExpressionAttributeValues": {":q": {"N":"1"}}
            }},
        ],
        ClientRequestToken = str(uuid.uuid4()),   # 10-min idempotency window
    )
except ClientError as e:
    if e.response["Error"]["Code"] == "TransactionCanceledException":
        # Inspect CancellationReasons[i].Code (None / ConditionalCheckFailed / ...)
        reasons = e.response["CancellationReasons"]
        raise ValueError(f"transaction rejected: {reasons}") from e
    raise

# End-to-end: order tracking app in ~30 lines.
# Single table "orders" with PK=CUSTOMER#, SK=ORDER#, GSI on status.
import boto3, uuid, datetime as dt
from boto3.dynamodb.conditions import Key
from decimal import Decimal

ddb   = boto3.resource("dynamodb", endpoint_url="http://localhost:8000",
                      region_name="us-east-1")
table = ddb.Table("orders")

def place_order(customer_id: str, total: Decimal) -> str:
    order_id = f"{dt.datetime.utcnow().isoformat(timespec='seconds')}Z-{uuid.uuid4().hex[:6]}"
    table.put_item(
        Item = {
            "pk":      f"CUSTOMER#{customer_id}",
            "sk":      f"ORDER#{order_id}",
            "status":  "pending",
            "created": order_id.split("-")[0],
            "total":   total,
        },
        ConditionExpression = "attribute_not_exists(sk)",
    )
    return order_id

def mark_paid(customer_id: str, order_id: str) -> dict:
    return table.update_item(
        Key = {"pk": f"CUSTOMER#{customer_id}", "sk": f"ORDER#{order_id}"},
        UpdateExpression   = "SET #s = :p",
        ConditionExpression = "#s = :pending",
        ExpressionAttributeNames  = {"#s": "status"},
        ExpressionAttributeValues = {":p":"paid", ":pending":"pending"},
        ReturnValues = "ALL_NEW",
    )["Attributes"]

def list_orders(customer_id: str) -> list:
    return table.query(
        KeyConditionExpression = Key("pk").eq(f"CUSTOMER#{customer_id}") &
                                 Key("sk").begins_with("ORDER#"),
        ScanIndexForward = False, Limit = 20,
    )["Items"]