def _get_tags_for_resource()

in source/code/services/rds_service.py [0:0]

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

    def _get_tags_for_resource(self, client, resource):
        """
        Returns the tags for specific resources that require additional boto calls to retrieve their tags.
        :param client: Client that can be used to make the boto call to retrieve the tags
        :param resource: The resource for which to retrieve the tags
        :return: Tags
        """

        # get the name of the proprty that holds the arn of the resource
        arn_property_name = "{}Arn".format(self._resource_name[0:-1])
        if arn_property_name[0:2].lower() == "db":
            arn_property_name = "DB{}".format(arn_property_name[2:])
        # get the arn of the resource
        resource_arn = resource[arn_property_name]

        # owner of the resource (could be other account for shared sbapshots)
        resource_owner_account = resource_arn.split(":")[4]
        resource_region = resource_arn.split(":")[3]

        if resource_owner_account == self.aws_account:
            # sane account, can use same session as used to retrieve the resource
            if self._use_cached_tags:
                self._tag_session = self.session

            # make sure the client has retries
            if getattr(self._service_client, "list_tags_for_resource" + boto_retry.DEFAULT_SUFFIX, None) is None:
                boto_retry.make_method_with_retries(boto_client_or_resource=client,
                                                    name="list_tags_for_resource",
                                                    service_retry_strategy=self._service_retry_strategy)
                self._tag_rds_client = client
        else:
            # resource is from other account, get a session to get the tags from that account as these are not
            # visible for shared rds resources
            if self._tag_account != resource_owner_account or self._tag_session is None:
                self._tag_account = resource_owner_account
                used_tag_role = None
                if self._tag_roles is not None:
                    # see if there is a role for the owner account
                    for role in self._tag_roles:
                        if role is not None and services.account_from_role_arn(role) == resource_owner_account:
                            used_tag_role = role
                            break
                    else:
                        # if there is no role and the account is the ops automator account use the default role
                        # in other cases it is not possible to retrieve the tags
                        if resource_owner_account != os.getenv(handlers.ENV_OPS_AUTOMATOR_ACCOUNT):
                            return {}
                self._tag_session = services.get_session(role_arn=used_tag_role)

            if not self._use_cached_tags:
                self._tag_rds_client = boto_retry.get_client_with_retries("rds", methods=["list_tags_for_resource"],
                                                                          context=self._context, region=resource_region)

        if self._use_cached_tags:
            return self.cached_tags(session=self._tag_session,
                                    resource_name=RESOURCES_WITH_TAGS[resource["ResourceTypeName"]],
                                    region=resource_region).get(resource_arn, {})

        try:
            resp = self._tag_rds_client.list_tags_for_resource_with_retries(ResourceName=resource_arn)
            return resp.get("TagList", [])
        except botocore.exceptions.ClientError as ex:
            if getattr(ex, "response", {}).get("Error", {}).get("Code", "") == "InvalidParameterValue":
                return []
            raise_exception("Can not list rds tags for resource {}, {}", resource_arn, ex)