如何在Python Lambda代码中引用CDK创建的SNS Topic ARN？

这是CDK开发里很常见的场景——因为资源ARN是部署时才生成的，没法提前硬编码到Lambda代码里。我平时做CDK开发经常遇到这种情况，给你两种最实用的解决思路：

方法1：通过环境变量传递ARN（最推荐） #

这是最简单直接的方式，CDK在部署时可以直接把生成的Topic ARN注入到Lambda的环境变量中，Lambda运行时读取这个变量即可。

第一步：修改CDK栈代码 #

在创建Lambda的时候，添加environment参数，把Topic的ARN传进去：

from aws_cdk import (
    aws_lambda as _lambda,
    aws_sns as sns,
    Stack,
)
from constructs import Construct

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

        # 创建SNS Topic
        my_topic = sns.Topic(self, "MyTopic")

        # 创建Lambda函数，传递Topic ARN到环境变量
        my_lambda = _lambda.Function(
            self, "MyLambdaFunction",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="lambda_function.lambda_handler",
            code=_lambda.Code.from_asset("path/to/your/lambda/code"),
            # 这里把Topic ARN注入环境变量
            environment={
                "SNS_TOPIC_ARN": my_topic.topic_arn
            }
        )

        # 给Lambda添加发布到Topic的权限
        my_topic.grant_publish(my_lambda)

第二步：修改Lambda运行时代码 #

在Python代码里读取环境变量，然后用它来调用SNS的publish方法：

import boto3
import os

sns = boto3.client('sns')

def lambda_handler(event, context):
    topic_arn = os.environ['SNS_TOPIC_ARN']
    response = sns.publish(
        TopicArn=topic_arn,
        Message="my_message"
    )
    return {"statusCode": 200, "body": response}

方法2：通过SSM Parameter Store存储ARN（适合跨栈或动态更新场景） #

如果你的Lambda需要跨栈访问这个Topic，或者后续可能要修改ARN而不想重新部署Lambda，可以把ARN存在SSM参数里，Lambda运行时读取参数。

第一步：CDK栈代码修改 #

先创建SSM参数存储Topic ARN，然后给Lambda添加读取SSM的权限：

from aws_cdk import (
    aws_lambda as _lambda,
    aws_sns as sns,
    aws_ssm as ssm,
    aws_iam as iam,
    Stack,
)
from constructs import Construct

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

        my_topic = sns.Topic(self, "MyTopic")

        # 把Topic ARN存入SSM参数
        ssm.StringParameter(self, "MyTopicArnParam",
            parameter_name="/myapp/sns/topic-arn",
            string_value=my_topic.topic_arn
        )

        my_lambda = _lambda.Function(
            self, "MyLambdaFunction",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="lambda_function.lambda_handler",
            code=_lambda.Code.from_asset("path/to/your/lambda/code")
        )

        # 给Lambda添加读取SSM参数的权限
        my_lambda.add_to_role_policy(iam.PolicyStatement(
            actions=["ssm:GetParameter"],
            resources=[ssm.StringParameter.from_string_parameter_name(self, "ParamArn", "/myapp/sns/topic-arn").parameter_arn]
        ))

        # 给Lambda添加发布到Topic的权限
        my_topic.grant_publish(my_lambda)

第二步：Lambda运行时代码修改 #

从SSM读取参数后调用SNS：

import boto3

sns = boto3.client('sns')
ssm = boto3.client('ssm')

def lambda_handler(event, context):
    # 读取SSM参数
    response = ssm.get_parameter(Name='/myapp/sns/topic-arn')
    topic_arn = response['Parameter']['Value']
    
    # 发布消息
    sns_response = sns.publish(
        TopicArn=topic_arn,
        Message="my_message"
    )
    return {"statusCode": 200, "body": sns_response}

小提醒 #

• 方法1是最常用的，因为不需要额外的API调用，性能更好，配置也简单。
• 不管用哪种方法，都要确保Lambda已经被授予了向该SNS Topic发布消息的权限（你已经配置了这部分，没问题）。

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