如何使用ResourceGroupsTaggingAPI获取AWS所有区域带指定标签的资源ARN

你当前返回结果不全是因为Resource Groups Tagging API是区域级服务，每个客户端默认只会请求初始化时指定的区域（未显式指定则使用本地配置的默认区域），仅能返回该区域内的匹配资源。要获取全区域的匹配ARN，只需遍历所有账号已启用的区域，为每个区域单独创建API客户端发起请求即可。

代码修改方案 #

核心修改逻辑如下：

• 新增获取账号全量已启用区域的逻辑，无需硬编码区域列表，自动适配账号实际开通的区域
• 外层新增区域遍历逻辑，每个区域单独初始化Tagging API客户端执行查询
• 全局资源（IAM、CloudFront、Route53等）的标签默认存储在us-east-1区域，遍历过程会自动覆盖，无需额外处理

修改后可运行代码：

import boto3
import csv

class TagEditor:
    
    def writeToCsv(self,writer, tag_list):
        for resource in tag_list:
            for tag in resource['Tags']:
                row = dict(
                    ResourceArn=resource['ResourceARN'], TagKey=tag['Key'], TagValue=tag['Value'])
                writer.writerow(row)
            
    def query_region_tags(self, restag, writer):
        # 过滤tag1
        response_deployment = restag.get_resources(TagFilters=[{'Key':'tag1'}],ResourcesPerPage=100)
        self.writeToCsv(writer, response_deployment['ResourceTagMappingList'])
        
        while 'PaginationToken' in response_deployment and response_deployment['PaginationToken']:
            token = response_deployment['PaginationToken']
            response_deployment = restag.get_resources(TagFilters=[{'Key':'tag1'}], ResourcesPerPage=100, PaginationToken=token)
            self.writeToCsv(writer, response_deployment['ResourceTagMappingList'])

        # 过滤tag2
        response_deployment = restag.get_resources(TagFilters=[{'Key':'tag2'}],ResourcesPerPage=100)
        self.writeToCsv(writer, response_deployment['ResourceTagMappingList'])
        
        while 'PaginationToken' in response_deployment and response_deployment['PaginationToken']:
            token = response_deployment['PaginationToken']
            response_deployment = restag.get_resources(TagFilters=[{'Key':'tag2'}], ResourcesPerPage=100, PaginationToken=token)
            self.writeToCsv(writer, response_deployment['ResourceTagMappingList'])
    
    def main(self):
        field_names = ['ResourceArn', 'TagKey', 'TagValue']
        s3 = boto3.client('s3')
        # 获取账号下所有已启用的区域
        ec2 = boto3.client('ec2', region_name='us-east-1')
        regions = [region['RegionName'] for region in ec2.describe_regions()['Regions']]
        
        with open('/tmp/tags.csv', 'w') as csvfile:
            writer = csv.DictWriter(csvfile, quoting=csv.QUOTE_ALL,
                                    delimiter=',', dialect='excel', fieldnames=field_names)
            writer.writeheader()
            
            # 遍历每个区域执行查询
            for region in regions:
                print(f"正在查询区域：{region}")
                try:
                    restag = boto3.client('resourcegroupstaggingapi', region_name=region)
                    self.query_region_tags(restag, writer)
                except Exception as e:
                    print(f"区域{region}查询失败：{str(e)}")
                    continue
        
        s3.upload_file('/tmp/tags.csv', Bucket = 'my-bucket', Key = 'tags.csv')

注意事项 #

• 请确保你使用的IAM身份（访问密钥/角色）具备所有区域的tag:GetResources权限，以及目标存储桶的s3:PutObject权限
• 如果不需要遍历所有区域，可自行在获取到的regions列表中过滤所需查询的区域
• ResourcesPerPage参数最高支持设置为100，可根据实际场景调整

内容的提问来源于stack exchange，提问作者Omega