使用boto3调用import_table报错：'DynamoDB'对象无该属性

报错核心原因 #

该报错是因为你的boto3版本过低，import_table对应DynamoDB的ImportTable API，较旧版本的boto3未集成该接口。

解决步骤 #

• 升级boto3和botocore到最新版本：

  pip install --upgrade boto3 botocore
  

• 验证版本兼容性：执行以下代码确认版本是否支持import_table（需至少1.20.0以上版本）

  import boto3
  print(boto3.__version__)
  

针对定期导入场景的额外优化 #

你提到会在导入前删除旧表，注意DynamoDB删除表是异步操作，直接删除后立即调用import_table会因表处于删除中状态报错，需添加等待表删除完成的逻辑。

修改后的完整代码示例 #

import boto3
from botocore.exceptions import ClientError
import time

class YourDataImporter:
    def __init__(self):
        self.session = boto3.Session()
        self.db_client = self.session.client("dynamodb")
        self.table_name = "your-target-table"
        self.bucket_name = "your-s3-bucket"
        self.report_csv_key = "path/to/your.csv"
        self.table_key_arn = "your-kms-key-arn"
        self.logger = ...  # 替换为你的logger实例

    def wait_for_table_deletion(self):
        """等待表完全删除"""
        while True:
            try:
                self.db_client.describe_table(TableName=self.table_name)
                self.logger.info(f"表 {self.table_name} 仍在删除中，等待10秒...")
                time.sleep(10)
            except ClientError as e:
                if e.response['Error']['Code'] == 'ResourceNotFoundException':
                    self.logger.info(f"表 {self.table_name} 已删除完成")
                    break
                else:
                    self.logger.error(f"检查表状态失败: {str(e)}")
                    raise

    def delete_old_table(self):
        """删除旧表（不存在则跳过）"""
        try:
            self.db_client.delete_table(TableName=self.table_name)
            self.logger.info(f"已触发删除表 {self.table_name}")
            self.wait_for_table_deletion()
        except ClientError as e:
            if e.response['Error']['Code'] == 'ResourceNotFoundException':
                self.logger.info(f"表 {self.table_name} 不存在，无需删除")
            else:
                self.logger.error(f"删除表失败: {str(e)}")
                raise

    def create_table_from_csv(self) -> dict:
        """从S3 CSV导入到DynamoDB新表"""
        response = None
        try:
            # 先清理旧表
            self.delete_old_table()
            
            self.logger.info(f"启动S3 CSV导入任务，目标表: {self.table_name}")
            response = self.db_client.import_table(
                S3BucketSource={
                    "S3Bucket": self.bucket_name,
                    "S3KeyPrefix": self.report_csv_key,
                },
                InputFormat="CSV",
                TableCreationParameters={
                    "TableName": self.table_name,
                    "AttributeDefinitions": [
                        {"AttributeName": "instanceArn", "AttributeType": "S"},
                        {"AttributeName": "instanceName", "AttributeType": "S"},
                    ],
                    "KeySchema": [
                        {"AttributeName": "instanceArn", "KeyType": "HASH"},
                        {"AttributeName": "instanceName", "KeyType": "RANGE"},
                    ],
                    "BillingMode": "PROVISIONED",
                    "ProvisionedThroughput": {
                        "ReadCapacityUnits": 5,
                        "WriteCapacityUnits": 5,
                    },
                    "SSESpecification": {
                        "Enabled": True,
                        "SSEType": "KMS",
                        "KMSMasterKeyId": self.table_key_arn,
                    },
                },
            )
            self.logger.info(f"导入任务已启动，任务ARN: {response['ImportTableDescription']['ImportArn']}")
        except ClientError as error:
            self.logger.error(f"导入CSV失败，文件路径: {self.report_csv_key}")
            self.logger.error(f"错误详情: {str(error)}")
            raise

        return response

权限补充说明 #

确保执行代码的IAM角色拥有以下权限：

• DynamoDB相关：dynamodb:ImportTable、dynamodb:DeleteTable、dynamodb:DescribeTable
• S3相关：s3:GetObject、s3:ListBucket
• KMS相关（若使用加密）：kms:Encrypt、kms:Decrypt、kms:DescribeKey

内容的提问来源于stack exchange，提问作者Jack