AWS Interactive Python Script For Adding a Field to a Table

add_dynamodb_field.py

# Usage: python3 add_dynamodb_field.py

import boto3
import time
import sys

def get_epoch_timestamp():
    return int(time.time())

def prompt_user():
    print("🟢 DynamoDB Table Field Updater")
    table_name = input("Enter the DynamoDB table name: ").strip()
    field_name = input("Enter the new field name to add: ").strip()
    default_value = input("Enter the default value for the new field: ").strip()
    match_term = input("Enter the keyword to match in Name/Description for timestamp override: ").strip()

    # Try to convert numeric input where applicable
    if default_value.isdigit():
        default_value = int(default_value)
    elif default_value.lower() == "null":
        default_value = None

    return table_name, field_name, default_value, match_term

def update_items(table, field_name, default_value, match_term):
    scan_kwargs = {}
    updated_count = 0

    while True:
        response = table.scan(**scan_kwargs)
        items = response.get('Items', [])

        for item in items:
            name = item.get("Name", "")
            description = item.get("Description", "")

            key_fields = ["Id", "Rule_Id"]
            try:
                key = {k: item[k] for k in key_fields}
            except KeyError:
                print(f"⚠️  Skipping item with missing key(s): {item}")
                continue

            if match_term.lower() in name.lower() or match_term.lower() in description.lower():
                field_value = get_epoch_timestamp()
            else:
                field_value = default_value

            try:
                table.update_item(
                    Key=key,
                    UpdateExpression=f"SET {field_name} = :val",
                    ExpressionAttributeValues={':val': field_value}
                )
                updated_count += 1
            except Exception as e:
                print(f"❌ Error updating item {key}: {str(e)}")

        if 'LastEvaluatedKey' in response:
            scan_kwargs['ExclusiveStartKey'] = response['LastEvaluatedKey']
        else:
            break

    print(f"\n✅ Completed. {updated_count} items updated.")

def main():
    try:
        table_name, field_name, default_value, match_term = prompt_user()
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table(table_name)
        update_items(table, field_name, default_value, match_term)
    except KeyboardInterrupt:
        print("\n❌ Operation cancelled.")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ Unexpected error: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()