一、环境准备与依赖
安装依赖与环境检查
在开始前，确保你的开发环境具备 Python 3.7 及以上，并具备网络连接以便从 PyPI 安装依赖。本文围绕 Python 操作 DynamoDB 的 boto3 使用教程，重点放在实战场景与代码示例上。若尚未安装 boto3，可以通过以下命令快速搭建环境：
pip install boto3
安装完成后，可以通过简单的版本检查来确认安装成功：
import boto3
print(boto3.__version__)
提示：为避免不同项目之间的依赖混淆，建议使用虚拟环境（如 venv）来管理 Python 环境。
配置凭证与默认区域
要让 boto3 访问 DynamoDB，你需要提供 AWS 凭证和默认区域。常见做法包括使用环境变量、AWS 配置文件或显式传入会话参数。下面列出两种常用方式，便于在不同场景中快速切换：
# 环境变量方式
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_DEFAULT_REGION=us-east-1
# 配置文件方式（推荐用于多环境）
aws configure
# 按提示输入 Access Key、Secret、默认区域等信息
要点：凭证要拥有 DynamoDB 的访问权限，区域与目标表所在区域保持一致，以避免跨区域请求带来的延迟与失败风险。
二、连接 DynamoDB 与创建表
连接与初始化
使用 boto3.Session 可以在同一应用中灵活管理不同账号或区域的连接。初始化后，通过 dynamodb.Table 获取对具体表的操作入口。下面示例展示如何建立连接并指向名为 Products 的表。
import boto3
# 指定配置文件的 session，或直接省略以使用默认配置
session = boto3.Session(profile_name='dev', region_name='us-east-1')
dynamodb = session.resource('dynamodb')
table = dynamodb.Table('Products') # 指向目标表
要点：通过 resource 获取的对象更贴近 Python 化的对象操作，适合日常 CRUD 场景。
创建表（示例）
在 DynamoDB 中创建表通常需要设定主键及容量单位，下面给出一个最简的创建表代码示例。注意：生产环境中往往通过 CloudFormation、Terraform 等基础设施即代码工具来管理表结构。
import boto3dynamodb = boto3.client('dynamodb', region_name='us-east-1')
response = dynamodb.create_table(TableName='Products',AttributeDefinitions=[{'AttributeName': 'sku', 'AttributeType': 'S'},{'AttributeName': 'category', 'AttributeType': 'S'},],KeySchema=[{'AttributeName': 'sku', 'KeyType': 'HASH'},],GlobalSecondaryIndexes=[{'IndexName': 'CategoryIndex','KeySchema': [{'AttributeName': 'category', 'KeyType': 'HASH'}],'Projection': {'ProjectionType': 'ALL'},'ProvisionedThroughput': {'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}}],ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
)
print(response)
等待就绪：创建表后通常需要等待表进入 ACTIVE 状态，可以使用等待器提升脚本稳定性：
table_exists = boto3.client('dynamodb').get_waiter('table_exists')
table_exists.wait(TableName='Products')
三、基本操作：CRUD 与查询
创建与写入数据
通过 put_item 可以将一个字典对象写入 DynamoDB 表中。请将字段设计与表的主键、全局二级索引保持一致，以便后续的查询与更新。
item = {'sku': 'SKU12345','name': '无标签连帽衫','price': 29.99,'stock': 100,'category': 'apparel'
}
table.put_item(Item=item)
注意：若 Item 中已有同样主键的项，put_item 会覆盖原有项，请在设计时确保主键的唯一性。
读取数据
通过 get_item 按主键读取单条记录，适用于快速获取具体 SKU 的商品信息。获取到的 Item 位于返回值的 Item 键中。
response = table.get_item(Key={'sku': 'SKU12345'})
item = response.get('Item')
print(item)
更新数据
更新常见操作包括修改字段、增加或减少数值等。这里演示一个原子更新，将库存量递减一个单位。
table.update_item(Key={'sku': 'SKU12345'},UpdateExpression='SET stock = stock - :dec',ExpressionAttributeValues={':dec': 1},ConditionExpression='stock >= :dec'
)
原子性：使用 UpdateExpression 配合条件表达式可以确保在并发场景下的安全更新，避免超卖。
删除数据
删除单条记录可通过 delete_item 实现，注意主键必须提供，否则无法定位要删除的项。
table.delete_item(Key={'sku': 'SKU12345'})
查询与扫描
当需要基于非主键属性检索数据时，可以使用 query（基于索引的有条件查询）或 scan（全表扫描）。下例展示基于全局二级索引 CategoryIndex 的查询。
from boto3.dynamodb.conditions import Key
response = table.query(IndexName='CategoryIndex',KeyConditionExpression=Key('category').eq('apparel')
)
items = response.get('Items', [])
print(len(items))
注意：Query 要求在 GSI 的分区键上进行条件约束，ProjectionExpression 可以用来控制返回字段以降低带宽成本。
四、性能与成本优化
分区键设计与二级索引
良好的分区键设计能显著提升吞吐量与查询效率。应用中应尽量保证分区键的高基数、低冲突，并合理使用 全局二级索引（GSI） 来支撑不同查询模式。下面示例展示了如何基于类别字段创建 CategoryIndex，以便按类别快速聚合与筛选。
# 参考前面的创建表示例，已包含 CategoryIndex 的定义
# 查询示例（同上）
按需容量与预算控制
DynamoDB 提供 Provisoned（有明确读写单位）与 On-Demand（按需）两种容量模式。对于流量波动大的场景，On-Demand 可以避免容量管理的复杂性；而稳定流量场景则可以通过设置更精准的 ReadCapacityUnits 与 WriteCapacityUnits 来降低成本。
# 使用 boto3 修改表的容量模式（示例：切换到 On-Demand）
dynamodb = boto3.client('dynamodb', region_name='us-east-1')
dynamodb.update_table(TableName='Products',BillingMode='PAY_PER_REQUEST'
)
五、错误处理与重试策略
常见异常处理
在生产环境中，网络波动和并发冲突常见，需要对 ClientError 以及具体的错误码做合理处理，以保持服务的鲁棒性。
import botocore
from botocore.exceptions import ClientErrortry:table.get_item(Key={'sku': 'SKU12345'})
except ClientError as e:error_code = e.response['Error']['Code']print('DynamoDB error:', error_code, e.response['Error']['Message'])
重试策略示例
对瞬时失败可实现简单的指数退避重试，避免对 DynamoDB 造成额外压力。以下示例演示一个简易的重试框架，辅以条件检查。
import time
from botocore.exceptions import ClientErrordef safe_update_item(table, key, update_expr, expr_vals, max_retries=3):for attempt in range(max_retries):try:table.update_item(Key=key,UpdateExpression=update_expr,ExpressionAttributeValues=expr_vals)return Trueexcept ClientError as e:if e.response['Error']['Code'] == 'ConditionalCheckFailedException':time.sleep(2 ** attempt)continueraisereturn False
六、实战案例：简单库存管理系统
设计目标与核心流程
本案例以一个简易的库存管理场景为例，演示如何通过 boto3 对 DynamoDB 进行库存扣减、下单与并发保护。核心流程包括：查询当前库存 -> 计算新库存 -> 原子更新库存 -> 处理并发冲突。
import boto3
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key# 假设已连接到 Products 表的 table 对象
def place_order(sku, qty, table):resp = table.get_item(Key={'sku': sku})item = resp.get('Item')if not item or item.get('stock', 0) < qty:raise ValueError('库存不足')new_stock = item['stock'] - qtytry:table.update_item(Key={'sku': sku},UpdateExpression='SET stock = :new',ConditionExpression='stock = :old',ExpressionAttributeValues={':new': new_stock,':old': item['stock']})return Trueexcept ClientError as e:if e.response['Error']['Code'] == 'ConditionalCheckFailedException':# 冲突：重新尝试下单return place_order(sku, qty, table)else:raise# 使用示例（确保已创建并获取到了 table 实例）
# place_order('SKU12345', 2, table)
要点：通过乐观锁与条件更新实现并发保护，确保同一时间不会超卖，从而提升系统的健壮性。该案例可以作为实际业务中库存扣减的模板。
