如何获取AWS CDK创建的DynamoDB表名称（遵循最佳实践）

你不需要硬编码表名，利用AWS Solutions Constructs的公开属性就能动态获取创建的资源，完全符合CDK最佳实践。

步骤说明 #

1. 创建IotToLambdaToDynamoDB构造实例时，不要硬编码表名（移除dynamo_table_props里的table_name，或通过栈名等动态生成）
2. 将构造实例赋值给变量，通过构造的lambda_function和dynamo_table属性访问对应的资源对象
3. 调用Lambda的add_environment方法，将动态获取的表名传入环境变量

修正后的完整代码 #

from aws_solutions_constructs.aws_iot_lambda_dynamodb import IotToLambdaToDynamoDB
from aws_cdk import (
    aws_iot as iot,
    aws_lambda as _lambda,
    aws_dynamodb as ddb,
    Stack
)
from constructs import Construct

class MyIotStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # 初始化构造实例，不硬编码表名
        iot_ddb_construct = IotToLambdaToDynamoDB(
            self,
            "test-iot-lambda-dynamodb-stack",
            lambda_function_props=_lambda.FunctionProps(
                code=_lambda.Code.from_asset("lambda"),
                runtime=_lambda.Runtime.PYTHON_3_9,
                handler="index.handler"
            ),
            iot_topic_rule_props=iot.CfnTopicRuleProps(
                topic_rule_payload=iot.CfnTopicRule.TopicRulePayloadProperty(
                    rule_disabled=False,
                    description="Processing of DTC messages from the AWS Connected Vehicle Solution.",
                    sql="SELECT * FROM 'connectedcar/dtc/#'",
                    actions=[]
                )
            ),
            dynamo_table_props=ddb.TableProps(
                partition_key={"name": "id", "type": ddb.AttributeType.STRING}
                # 可选：如果需要自定义表名，用动态方式生成，避免硬编码重复
                # table_name=f"{self.stack_name}-dtc-events"
            )
        )

        # 动态获取表名，添加到Lambda环境变量
        iot_ddb_construct.lambda_function.add_environment(
            "TABLE_NAME",
            iot_ddb_construct.dynamo_table.table_name
        )

Lambda代码示例（index.py） #

在Lambda中直接读取环境变量即可使用表名：

import boto3
import os

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])

def handler(event, context):
    # 处理IoT消息并写入DynamoDB
    table.put_item(
        Item={
            "id": event["id"],
            "payload": event,
            "timestamp": event.get("timestamp")
        }
    )
    return {"statusCode": 200, "message": "Record written successfully"}

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