Python中如何从AWS CDK向外部文件动态传递DynamoDB表名

以下三个方案按实现难度从低到高排列，可根据你的使用场景选择：

方案1：抽离公共常量（最适合纯固定表名场景） #

如果你的表名不需要动态生成（比如不需要加环境、日期后缀），直接把表名抽成公共常量是最简单的实现方式：

• 新建constants.py文件，写入公共常量：

DYNAMO_TABLE_NAME = "DynamoTableToPassInOtherFile"

• 修改第一个CDK文件，导入常量使用（补全原代码缺失的date导入）：

from aws_cdk import (
core as cdk,
aws_dynamodb as dynamodb
)
from datetime import date
from constants import DYNAMO_TABLE_NAME

class CdkStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        currentDate= date.today()
        dynamodbTable= dynamodb.Table(self,id='dynamodbTable',table_name=DYNAMO_TABLE_NAME,partition_key=dynamodb.Attribute(name='id',type=dynamodb.AttributeType.STRING))

• 修改第二个Python文件，导入常量替换硬编码的表名：

from constants import DYNAMO_TABLE_NAME
import boto3

dynamodb_client = boto3.client('dynamodb')

def write_to_dynamo_db(self, json_content):
    dynamodb_response = dynamodb_client.put_item(
                TableName= DYNAMO_TABLE_NAME,
                Item={
                    'id':{'S': self.id},
                    'formatted_date': {'S': json_content},
                    },
                ReturnValues='ALL_OLD',
                ReturnConsumedCapacity='INDEXES',
            )
    return dynamodb_response

方案2：CDK导出输出到配置文件（适合本地运行第二个文件的场景） #

如果你的表名是动态生成的（比如加了日期、环境后缀），可以通过CDK的输出功能把表名导出到本地配置文件，第二个文件直接读取即可：

• 修改第一个CDK文件，添加输出配置：

from aws_cdk import (
core as cdk,
aws_dynamodb as dynamodb
)
from datetime import date

class CdkStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        currentDate= date.today()
        table_name= 'DynamoTableToPassInOtherFile'
        dynamodbTable= dynamodb.Table(self,id='dynamodbTable',table_name=table_name,partition_key=dynamodb.Attribute(name='id',type=dynamodb.AttributeType.STRING))
        # 新增输出项
        core.CfnOutput(self, 
            id="DynamoTableNameOutput",
            value=dynamodbTable.table_name,
            export_name="DynamoTableName"
        )

• 部署CDK时执行命令，把输出内容保存到本地json文件：
  cdk deploy --outputs-file cdk-outputs.json
• 修改第二个Python文件，读取配置文件获取表名：

import json
import boto3

dynamodb_client = boto3.client('dynamodb')

def write_to_dynamo_db(self, json_content):
    # 读取CDK输出文件
    with open("cdk-outputs.json", "r", encoding="utf-8") as f:
        cdk_outputs = json.load(f)
    # 占位符替换为你实际的CDK栈名
    table_name = cdk_outputs["你的栈名"]["DynamoTableNameOutput"]
    dynamodb_response = dynamodb_client.put_item(
                TableName= table_name,
                Item={
                    'id':{'S': self.id},
                    'formatted_date': {'S': json_content},
                    },
                ReturnValues='ALL_OLD',
                ReturnConsumedCapacity='INDEXES',
            )
    return dynamodb_response

方案3：存储到SSM参数存储（适合第二个文件在AWS云环境运行的场景） #

如果第二个文件是在ECS、Lambda等AWS云资源上运行，可以把表名存到SSM参数存储，运行时动态获取，避免本地配置文件的同步问题：

• 修改第一个CDK文件，添加SSM参数配置：

from aws_cdk import (
core as cdk,
aws_dynamodb as dynamodb,
aws_ssm as ssm
)
from datetime import date

class CdkStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        currentDate= date.today()
        table_name= 'DynamoTableToPassInOtherFile'
        dynamodbTable= dynamodb.Table(self,id='dynamodbTable',table_name=table_name,partition_key=dynamodb.Attribute(name='id',type=dynamodb.AttributeType.STRING))
        # 新增SSM参数存储配置
        ssm.StringParameter(self,
            id="DynamoTableNameParam",
            parameter_name="/config/dynamo/table_name",
            string_value=dynamodbTable.table_name
        )

• 修改第二个Python文件，调用SSM接口获取表名：

import boto3

ssm_client = boto3.client('ssm')
dynamodb_client = boto3.client('dynamodb')
# 全局缓存表名，避免每次调用都请求SSM
_TABLE_NAME = None

def write_to_dynamo_db(self, json_content):
    global _TABLE_NAME
    if not _TABLE_NAME:
        resp = ssm_client.get_parameter(Name="/config/dynamo/table_name")
        _TABLE_NAME = resp["Parameter"]["Value"]
    dynamodb_response = dynamodb_client.put_item(
                TableName= _TABLE_NAME,
                Item={
                    'id':{'S': self.id},
                    'formatted_date': {'S': json_content},
                    },
                ReturnValues='ALL_OLD',
                ReturnConsumedCapacity='INDEXES',
            )
    return dynamodb_response

注意：使用该方案需要给运行第二个文件的IAM角色添加ssm:GetParameter权限，同时需要保证角色有DynamoDB的PutItem权限。

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