You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用Python+moto 5本地模拟AWS DynamoDB并解决认证错误

解决方案:持久化本地DynamoDB模拟类(基于boto3+moto)

以下是符合需求的实现代码,解决了表持久化和GetRoleCredentials错误的问题:

import boto3
from moto import mock_dynamodb

class LocalDynamoDB:
    def __init__(self):
        # 启动全局DynamoDB Mock,生命周期与类实例绑定
        self.mock = mock_dynamodb()
        self.mock.start()
        
        # 使用静态虚拟凭证,避免触发真实AWS凭证请求
        self.dynamodb = boto3.resource(
            'dynamodb',
            region_name='us-east-1',
            aws_access_key_id='dummy_key',
            aws_secret_access_key='dummy_secret'
        )

    def create_table(self, table_name, key_schema, attribute_definitions, provisioned_throughput):
        return self.dynamodb.create_table(
            TableName=table_name,
            KeySchema=key_schema,
            AttributeDefinitions=attribute_definitions,
            ProvisionedThroughput=provisioned_throughput
        )

    def put_item(self, table_name, item):
        table = self.dynamodb.Table(table_name)
        return table.put_item(Item=item)

    def get_item(self, table_name, key):
        table = self.dynamodb.Table(table_name)
        return table.get_item(Key=key)

    def stop(self):
        # 可选:程序退出时停止Mock
        self.mock.stop()

关键说明

  • 持久化Mock环境:在类初始化时启动全局mock_dynamodb并持有实例,而非单独装饰每个方法。这样整个程序运行期间Mock环境不会重建,创建的表会一直保留。
  • 解决凭证错误:显式设置虚拟的aws_access_key_id和aws_secret_access_key,阻止boto3尝试请求真实AWS凭证,从而避免触发未实现的GetRoleCredentials接口调用。
  • 兼容真实接口:所有操作方法直接调用boto3的DynamoDB资源接口,和真实环境的使用方式完全一致,业务代码无需额外修改。

使用示例

# 初始化本地DynamoDB实例
local_ddb = LocalDynamoDB()

# 创建测试表
local_ddb.create_table(
    table_name='user_table',
    key_schema=[{'AttributeName': 'user_id', 'KeyType': 'HASH'}],
    attribute_definitions=[{'AttributeName': 'user_id', 'AttributeType': 'S'}],
    provisioned_throughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
)

# 插入数据
local_ddb.put_item('user_table', {'user_id': 'u001', 'username': 'test_user'})

# 查询数据
response = local_ddb.get_item('user_table', {'user_id': 'u001'})
print(response['Item'])

# 程序结束时可选停止Mock
# local_ddb.stop()

内容的提问来源于stack exchange,提问作者TheMemeMachine

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.17 05:21:05