in chalice/deploy/planner.py [0:0]
def _plan_snslambdasubscription(self, resource):
# type: (models.SNSLambdaSubscription) -> Sequence[InstructionMsg]
function_arn = Variable(
'%s_lambda_arn' % resource.lambda_function.resource_name
)
topic_arn_varname = '%s_topic_arn' % resource.resource_name
subscribe_varname = '%s_subscription_arn' % resource.resource_name
instruction_for_topic_arn = [] # type: List[InstructionMsg]
if re.match(r"^arn:aws[a-z\-]*:sns:", resource.topic):
instruction_for_topic_arn += [
models.StoreValue(
name=topic_arn_varname,
value=resource.topic,
)
]
else:
# To keep the user API simple, we only require the topic
# name and not the ARN. However, the APIs require the topic
# ARN so we need to reconstruct it here in the planner.
instruction_for_topic_arn += self._arn_parse_instructions(
function_arn) + [
models.StoreValue(
name=topic_arn_varname,
value=StringFormat(
'arn:{partition}:sns:{region_name}:{account_id}:%s' % (
resource.topic
),
['partition', 'region_name', 'account_id'],
),
)
]
if self._remote_state.resource_exists(resource):
# Given there's nothing about an SNS subscription you can
# configure for now, if the resource exists, we don't do
# anything. The resource sweeper will verify that if the
# subscription doesn't actually apply that we should unsubscribe
# from the topic.
deployed = self._remote_state.resource_deployed_values(resource)
subscription_arn = deployed['subscription_arn']
return instruction_for_topic_arn + self._batch_record_resource(
'sns_event', resource.resource_name, {
'topic': resource.topic,
'lambda_arn': Variable(function_arn.name),
'subscription_arn': subscription_arn,
'topic_arn': Variable(topic_arn_varname),
}
)
return instruction_for_topic_arn + [
models.APICall(
method_name='add_permission_for_sns_topic',
params={'topic_arn': Variable(topic_arn_varname),
'function_arn': function_arn},
),
(models.APICall(
method_name='subscribe_function_to_topic',
params={'topic_arn': Variable(topic_arn_varname),
'function_arn': function_arn},
output_var=subscribe_varname,
), 'Subscribing %s to SNS topic %s\n'
% (resource.lambda_function.function_name, resource.topic)
)
] + self._batch_record_resource(
'sns_event', resource.resource_name, {
'topic': resource.topic,
'lambda_arn': Variable(function_arn.name),
'subscription_arn': Variable(subscribe_varname),
'topic_arn': Variable(topic_arn_varname),
}
)