def _construct_event_invoke_config()

in samtranslator/model/sam_resources.py [0:0]

• 47 lines of code
• 11 McCabe index (conditional complexity)

    def _construct_event_invoke_config(self, function_name, alias_name, lambda_alias, intrinsics_resolver, conditions):
        """
        Create a `AWS::Lambda::EventInvokeConfig` based on the input dict `EventInvokeConfig`
        """
        resources = []
        policy_document = []

        # Try to resolve.
        resolved_event_invoke_config = intrinsics_resolver.resolve_parameter_refs(self.EventInvokeConfig)

        logical_id = "{id}EventInvokeConfig".format(id=function_name)
        if lambda_alias:
            lambda_event_invoke_config = LambdaEventInvokeConfig(
                logical_id=logical_id, depends_on=[lambda_alias.logical_id], attributes=self.resource_attributes
            )
        else:
            lambda_event_invoke_config = LambdaEventInvokeConfig(
                logical_id=logical_id, attributes=self.resource_attributes
            )

        dest_config = {}
        input_dest_config = resolved_event_invoke_config.get("DestinationConfig")
        if input_dest_config and input_dest_config.get("OnSuccess") is not None:
            resource, on_success, policy = self._validate_and_inject_resource(
                input_dest_config.get("OnSuccess"), "OnSuccess", logical_id, conditions
            )
            dest_config["OnSuccess"] = on_success
            self.EventInvokeConfig["DestinationConfig"]["OnSuccess"]["Destination"] = on_success.get("Destination")
            if resource is not None:
                resources.extend([resource])
            if policy is not None:
                policy_document.append(policy)

        if input_dest_config and input_dest_config.get("OnFailure") is not None:
            resource, on_failure, policy = self._validate_and_inject_resource(
                input_dest_config.get("OnFailure"), "OnFailure", logical_id, conditions
            )
            dest_config["OnFailure"] = on_failure
            self.EventInvokeConfig["DestinationConfig"]["OnFailure"]["Destination"] = on_failure.get("Destination")
            if resource is not None:
                resources.extend([resource])
            if policy is not None:
                policy_document.append(policy)

        lambda_event_invoke_config.FunctionName = ref(function_name)
        if alias_name:
            lambda_event_invoke_config.Qualifier = alias_name
        else:
            lambda_event_invoke_config.Qualifier = "$LATEST"
        lambda_event_invoke_config.DestinationConfig = dest_config
        lambda_event_invoke_config.MaximumEventAgeInSeconds = resolved_event_invoke_config.get(
            "MaximumEventAgeInSeconds"
        )
        lambda_event_invoke_config.MaximumRetryAttempts = resolved_event_invoke_config.get("MaximumRetryAttempts")
        resources.extend([lambda_event_invoke_config])

        return resources, policy_document