DynamoDB Hello World：学生表 CRUD 操作

目标

使用 Python (boto3) 操作 DynamoDB，完成建表、插入、查询、更新、删除等核心操作。

完整代码

# pip install boto3
import boto3
from decimal import Decimal
from datetime import datetime

# 使用本地 DynamoDB（开发环境）
dynamodb = boto3.resource(
    'dynamodb',
    endpoint_url='http://localhost:8000',
    region_name='us-east-1',
    aws_access_key_id='fake',
    aws_secret_access_key='fake'
)

# === 1. 创建表 ===
def create_table():
    table = dynamodb.create_table(
        TableName='Students',
        KeySchema=[
            {'AttributeName': 'student_id', 'KeyType': 'HASH'},  # 分区键
            {'AttributeName': 'enrolled_date', 'KeyType': 'RANGE'} # 排序键
        ],
        AttributeDefinitions=[
            {'AttributeName': 'student_id', 'AttributeType': 'S'},
            {'AttributeName': 'enrolled_date', 'AttributeType': 'S'}
        ],
        BillingMode='PAY_PER_REQUEST'  # 按需付费
    )
    table.wait_until_exists()
    print(f"表 {table.table_name} 已创建, ARN: {table.table_arn}")
    return table

# === 2. 插入数据 ===
def insert_student(table):
    items = [
        {
            'student_id': 'S2024001',
            'enrolled_date': '2024-09-01',
            'name': '张三',
            'age': 21,
            'major': '计算机科学',
            'gpa': Decimal('3.8'),
            'courses': {'数据库原理', '算法导论'}
        },
        {
            'student_id': 'S2024001',
            'enrolled_date': '2024-09-15',
            'name': '张三',
            'age': 21,
            'major': '计算机科学',
            'gpa': Decimal('3.9'),
            'courses': {'操作系统', '计算机网络'}
        },
        {
            'student_id': 'S2024002',
            'enrolled_date': '2024-09-01',
            'name': '李四',
            'age': 22,
            'major': '数学',
            'gpa': Decimal('3.5'),
            'courses': {'高等代数'}
        }
    ]
    
    for item in items:
        table.put_item(Item=item)
    print(f"插入 {len(items)} 条记录")

# === 3. 查询 ===
def query_students(table, student_id):
    """按分区键查询某学生的所有记录"""
    response = table.query(
        KeyConditionExpression=boto3.dynamodb.conditions.Key('student_id').eq(student_id)
    )
    print(f"\n查询 {student_id} 的记录:")
    for item in response['Items']:
        print(f"  日期: {item['enrolled_date']}, GPA: {item['gpa']}")

# === 4. 获取单条记录 ===
def get_item(table, student_id, enrolled_date):
    response = table.get_item(
        Key={
            'student_id': student_id,
            'enrolled_date': enrolled_date
        }
    )
    if 'Item' in response:
        item = response['Item']
        print(f"\n获取到: {item['name']}, GPA: {item['gpa']}")
    else:
        print("未找到记录")

# === 5. 更新记录 ===
def update_gpa(table, student_id, enrolled_date, new_gpa):
    response = table.update_item(
        Key={
            'student_id': student_id,
            'enrolled_date': enrolled_date
        },
        UpdateExpression='SET gpa = :gpa, #ts = :ts',
        ExpressionAttributeValues={
            ':gpa': Decimal(str(new_gpa)),
            ':ts': datetime.now().isoformat()
        },
        ExpressionAttributeNames={
            '#ts': 'updated_at'  # 新增字段
        },
        ReturnValues='UPDATED_NEW'
    )
    print(f"\n更新后: {response['Attributes']}")

# === 6. 使用 GSI（全局二级索引）按专业查询 ===
def create_gsi(table):
    """添加全局二级索引：按 major 查询"""
    table.update(
        AttributeDefinitions=[
            {'AttributeName': 'major', 'AttributeType': 'S'}
        ],
        GlobalSecondaryIndexUpdates=[
            {
                'Create': {
                    'IndexName': 'MajorIndex',
                    'KeySchema': [
                        {'AttributeName': 'major', 'KeyType': 'HASH'},
                        {'AttributeName': 'gpa', 'KeyType': 'RANGE'}
                    ],
                    'Projection': {'ProjectionType': 'ALL'}
                }
            }
        ]
    )
    print("GSI MajorIndex 创建中（约 1 分钟）...")

# === 主流程 ===
if __name__ == '__main__':
    try:
        table = create_table()
    except Exception:
        table = dynamodb.Table('Students')
        print("表已存在")

    insert_student(table)
    query_students(table, 'S2024001')
    get_item(table, 'S2024001', '2024-09-01')
    update_gpa(table, 'S2024002', '2024-09-01', 3.7)

预期输出

表 Students 已创建
插入 3 条记录

查询 S2024001 的记录:
  日期: 2024-09-01, GPA: 3.8
  日期: 2024-09-15, GPA: 3.9

获取到: 张三, GPA: 3.8

更新后: {'gpa': Decimal('3.7'), 'updated_at': '2024-...'}

关键点

• DynamoDB 的 PRIMARY KEY = Partition Key (HASH) + 可选的 Sort Key (RANGE)
• query() 需要完整 Partition Key，scan() 全表扫描（较慢）
• GSI 允许按非主键字段查询，但最终一致
• Decimal 类型用于精确数值（boto3 要求）
• 集合类型（如 courses）支持 {'数据库原理', '算法导论'}