DynamoDB Basics Explained: Tables, Keys, and Real-World Patterns

import boto3

# Create a DynamoDB client pointing at a local DynamoDB instance for dev
# In production, remove 'endpoint_url' and use your AWS region
dynamodb = boto3.resource(
    'dynamodb',
    region_name='us-east-1',
    endpoint_url='http://localhost:8000'  # local DynamoDB for safe testing
)

# Create the Orders table with a COMPOSITE primary key:
#   partition_key = customer_id  (groups all orders for one customer together)
#   sort_key      = order_date#order_id  (allows range queries by date)
table = dynamodb.create_table(
    TableName='Orders',
    KeySchema=[
        {
            'AttributeName': 'customer_id',   # partition key
            'KeyType': 'HASH'
        },
        {
            'AttributeName': 'order_date_order_id',  # sort key (composite value)
            'KeyType': 'RANGE'
        }
    ],
    AttributeDefinitions=[
        # Only KEY attributes go here — DynamoDB is schemaless for everything else
        {'AttributeName': 'customer_id',        'AttributeType': 'S'},  # S = String
        {'AttributeName': 'order_date_order_id','AttributeType': 'S'}
    ],
    BillingMode='PAY_PER_REQUEST'  # on-demand pricing — no capacity planning needed for dev
)

# Wait until the table is actually ready before writing to it
table.wait_until_exists()
print(f"Table status: {table.table_status}")
print(f"Table name:   {table.table_name}")

import boto3
from boto3.dynamodb.conditions import Key
from decimal import Decimal  # DynamoDB uses Decimal, not float, for numbers

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

# ─── WRITE: Put three orders for the same customer ─────────────────────────
# Sort key format: "YYYY-MM-DD#ORDER-ID" — date first so range queries work
orders_to_insert = [
    {
        'customer_id':         'CUST-42',
        'order_date_order_id': '2024-01-15#ORD-001',
        'total_usd':           Decimal('59.99'),
        'status':              'DELIVERED',
        'items':               ['Mechanical Keyboard', 'USB Hub']
    },
    {
        'customer_id':         'CUST-42',
        'order_date_order_id': '2024-03-22#ORD-002',
        'total_usd':           Decimal('149.00'),
        'status':              'SHIPPED',
        'items':               ['Monitor Stand']
    },
    {
        'customer_id':         'CUST-42',
        'order_date_order_id': '2024-06-10#ORD-003',
        'total_usd':           Decimal('29.99'),
        'status':              'PROCESSING',
        'items':               ['Desk Mat']
    }
]

for order in orders_to_insert:
    table.put_item(Item=order)  # put_item: create or fully overwrite an item

print("Inserted 3 orders for CUST-42")

# ─── READ: Get ONE specific order by its full primary key ───────────────────
response = table.get_item(
    Key={
        'customer_id':         'CUST-42',
        'order_date_order_id': '2024-03-22#ORD-002'
    }
)
specific_order = response.get('Item', {})
print(f"\nSpecific order: {specific_order['order_date_order_id']} — ${specific_order['total_usd']}")

# ─── QUERY: Get all orders for CUST-42 placed on or after 2024-03-01 ────────
# 'between' on the sort key stays inside one partition — this is efficient!
range_response = table.query(
    KeyConditionExpression=
        Key('customer_id').eq('CUST-42') &
        Key('order_date_order_id').between('2024-03-01', '2024-12-31')
)

print(f"\nOrders from March 2024 onwards ({range_response['Count']} found):")
for order in range_response['Items']:
    print(f"  {order['order_date_order_id']} | ${order['total_usd']} | {order['status']}")

# ─── UPDATE: Change just the status field without rewriting the whole item ───
table.update_item(
    Key={
        'customer_id':         'CUST-42',
        'order_date_order_id': '2024-03-22#ORD-002'
    },
    UpdateExpression='SET #s = :new_status',  # #s avoids 'status' reserved word conflict
    ExpressionAttributeNames={'#s': 'status'},
    ExpressionAttributeValues={':new_status': 'DELIVERED'}
)
print("\nUpdated ORD-002 status to DELIVERED")

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource(
    'dynamodb',
    region_name='us-east-1',
    endpoint_url='http://localhost:8000'
)

# Add a GSI to the existing Orders table so we can query by order status
# GSI partition key: status  |  GSI sort key: order_date_order_id
# This lets us ask: "give me all SHIPPED orders, most recent first"
dynamodb.meta.client.update_table(
    TableName='Orders',
    AttributeDefinitions=[
        {'AttributeName': 'status',              'AttributeType': 'S'},
        {'AttributeName': 'order_date_order_id', 'AttributeType': 'S'}
    ],
    GlobalSecondaryIndexUpdates=[
        {
            'Create': {
                'IndexName': 'StatusByDate-GSI',
                'KeySchema': [
                    {'AttributeName': 'status',              'KeyType': 'HASH'},
                    {'AttributeName': 'order_date_order_id', 'KeyType': 'RANGE'}
                ],
                'Projection': {
                    # INCLUDE only the fields we actually need — saves cost vs ALL
                    'ProjectionType': 'INCLUDE',
                    'NonKeyAttributes': ['customer_id', 'total_usd']
                }
            }
        }
    ]
)

print("GSI creation initiated (takes a moment to backfill)...")

# Wait for the GSI to become ACTIVE before querying it
waiter = dynamodb.meta.client.get_waiter('table_exists')
waiter.wait(TableName='Orders')

table = dynamodb.Table('Orders')

# ─── Query via GSI: all SHIPPED orders ──────────────────────────────────────
gsi_response = table.query(
    IndexName='StatusByDate-GSI',   # tell DynamoDB to use our new index
    KeyConditionExpression=Key('status').eq('SHIPPED')
)

print(f"\nAll SHIPPED orders ({gsi_response['Count']} found):")
for order in gsi_response['Items']:
    print(f"  Customer: {order['customer_id']} | "
          f"Order: {order['order_date_order_id']} | "
          f"Total: ${order['total_usd']}")

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

dynamodb = boto3.resource(
    'dynamodb',
    region_name='us-east-1',
    endpoint_url='http://localhost:8000'
)

# One table to rule them all — stores Customers AND Orders together
# Generic key names PK / SK are conventional in single-table design
table = dynamodb.create_table(
    TableName='ECommerceApp',
    KeySchema=[
        {'AttributeName': 'PK', 'KeyType': 'HASH'},
        {'AttributeName': 'SK', 'KeyType': 'RANGE'}
    ],
    AttributeDefinitions=[
        {'AttributeName': 'PK', 'AttributeType': 'S'},
        {'AttributeName': 'SK', 'AttributeType': 'S'}
    ],
    BillingMode='PAY_PER_REQUEST'
)
table.wait_until_exists()

# ─── Write a CUSTOMER record ─────────────────────────────────────────────────
# PK and SK both describe what the item IS — the value encodes entity type
table.put_item(Item={
    'PK':           'CUSTOMER#CUST-42',   # partition key identifies the customer
    'SK':           'PROFILE',            # sort key distinguishes record type
    'entity_type':  'Customer',           # explicit type tag — helps with filtering
    'full_name':    'Alex Rivera',
    'email':        'alex@example.com',
    'loyalty_tier': 'Gold'
})

# ─── Write two ORDERS for the same customer ──────────────────────────────────
# Same PK as the customer — they share a partition, so one Query fetches both
table.put_item(Item={
    'PK':          'CUSTOMER#CUST-42',
    'SK':          'ORDER#2024-01-15#ORD-001',   # ORDER# prefix separates orders from profile
    'entity_type': 'Order',
    'total_usd':   Decimal('59.99'),
    'status':      'DELIVERED'
})
table.put_item(Item={
    'PK':          'CUSTOMER#CUST-42',
    'SK':          'ORDER#2024-06-10#ORD-003',
    'entity_type': 'Order',
    'total_usd':   Decimal('29.99'),
    'status':      'PROCESSING'
})

print("Inserted customer profile and 2 orders into single table")

# ─── Fetch customer profile + ALL orders in ONE query call ───────────────────
# begins_with('ORDER#') filters to just orders, skipping the PROFILE item
all_items_response = table.query(
    KeyConditionExpression=Key('PK').eq('CUSTOMER#CUST-42')
)
print(f"\nAll items for CUST-42 ({all_items_response['Count']} total):")
for item in all_items_response['Items']:
    print(f"  [{item['entity_type']}] SK={item['SK']}")

# Get ONLY the orders with a begins_with filter on the sort key
orders_only_response = table.query(
    KeyConditionExpression=
        Key('PK').eq('CUSTOMER#CUST-42') &
        Key('SK').begins_with('ORDER#')
)
print(f"\nOrders only ({orders_only_response['Count']} found):")
for order in orders_only_response['Items']:
    print(f"  {order['SK']} | ${order['total_usd']} | {order['status']}")

import boto3
import time
import random
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('Orders')

def put_item_with_retry(item, max_retries=3):
    """Put an item with exponential backoff and jitter.
    
    AWS SDK default retry is fine for background jobs, but for user-facing
    endpoints we want faster fallback and better logging.
    """
    for attempt in range(max_retries):
        try:
            table.put_item(Item=item)
            return True
        except ClientError as e:
            code = e.response['Error']['Code']
            if code == 'ProvisionedThroughputExceededException':
                # Exponential backoff with jitter: base = 0.1s, factor = 2, cap = 2s
                sleep_time = min(0.1 * (2 ** attempt) + random.random() * 0.1, 2.0)
                print(f"Retry {attempt+1} after {sleep_time:.2f}s — throttled")
                time.sleep(sleep_time)
            else:
                # Non-throttling error — re-raise immediately
                raise
    print("Failed after max retries — consider failing over or returning 503")
    return False

# Usage: put_item_with_retry({'customer_id': 'CUST-42', ...})

# ─── Full pagination example for Query ───────────────────────────────────────
def get_all_orders(customer_id):
    """Handle pagination correctly — never assume one page is all the data."""
    all_items = []
    last_key = None
    while True:
        query_kwargs = {
            'KeyConditionExpression': Key('customer_id').eq(customer_id)
        }
        if last_key:
            query_kwargs['ExclusiveStartKey'] = last_key
        
        response = table.query(**query_kwargs)
        all_items.extend(response['Items'])
        
        last_key = response.get('LastEvaluatedKey')
        if not last_key:
            break
    return all_items