def _link_policy()

in samtranslator/model/eventsources/pull.py [0:0]

• 22 lines of code
• 21 McCabe index (conditional complexity)

    def _link_policy(self, role, destination_config_policy=None):  # type: ignore[no-untyped-def]
        """If this source triggers a Lambda function whose execution role is auto-generated by SAM, add the
        appropriate managed policy to this Role.

        :param model.iam.IAMRole role: the execution role generated for the function
        """
        policy_arn = self.get_policy_arn()
        policy_statements = self.get_policy_statements()
        if role is not None:
            if policy_arn is not None and policy_arn not in role.ManagedPolicyArns:
                role.ManagedPolicyArns.append(policy_arn)
            if policy_statements is not None:
                if role.Policies is None:
                    role.Policies = []
                for policy in policy_statements:
                    if policy not in role.Policies and policy.get("PolicyDocument") not in [
                        d["PolicyDocument"] for d in role.Policies
                    ]:
                        role.Policies.append(policy)
        # add SQS or SNS policy only if role is present in kwargs
        if role is not None and destination_config_policy is not None and destination_config_policy:
            if role.Policies is None:
                role.Policies = []
                role.Policies.append(destination_config_policy)
            if role.Policies and destination_config_policy not in role.Policies:
                policy_document = destination_config_policy.get("PolicyDocument")
                # do not add the policy if the same policy document is already present
                if policy_document not in [d.get("PolicyDocument", {}) for d in role.Policies]:
                    role.Policies.append(destination_config_policy)