def evaluate()

in cfn_policy_validator/parsers/utils/intrinsic_functions/dynamic_ref_evaluator.py [0:0]

• 23 lines of code
• 1 McCabe index (conditional complexity)

	def evaluate(self, value):
		dynamic_ssm_references = dynamic_ssm_reference_regex.findall(value)
		for dynamic_ssm_reference in dynamic_ssm_references:
			dynamic_reference_text = dynamic_ssm_reference[0]
			parameter_name = dynamic_ssm_reference[1]
			parameter_version = dynamic_ssm_reference[2]

			if parameter_version == '':
				raise ApplicationError('Dynamic references to SSM parameters must include a version number to ensure the'
									   ' value does not change between validation and deployment.  Invalid dynamic '
									   f'reference: {dynamic_reference_text}')

			parameter_with_version = f'{parameter_name}:{parameter_version}'

			try:
				response = self.ssm_client.get_parameter(
					Name=parameter_with_version
				)
			except ClientError as e:
				if e.response['Error']['Code'] == 'ParameterNotFound':
					raise ApplicationError(f'Could not find SSM parameter referenced by dynamic reference: {dynamic_reference_text}')
				if e.response['Error']['Code'] == 'ParameterVersionNotFound':
					raise ApplicationError(f'Could not find version {parameter_version} of SSM parameter referenced by dynamic reference: {dynamic_reference_text}')
				raise

			value = value.replace(dynamic_ssm_reference[0], response['Parameter']['Value'])

		return value