Skip to content

Instantly share code, notes, and snippets.

@jesperalmstrom
Created February 12, 2025 07:55
Show Gist options
  • Save jesperalmstrom/f20b91d4b7ae891a6850e358687d1cd7 to your computer and use it in GitHub Desktop.
Save jesperalmstrom/f20b91d4b7ae891a6850e358687d1cd7 to your computer and use it in GitHub Desktop.
By using input parameters you will be able to change the tagging of all resources in a AWS account.
import argparse
import logging
import boto3
from botocore.exceptions import ClientError
def setup_logger(log_level):
"""
Configures and returns a logger instance.
"""
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
return logging.getLogger(__name__)
def chunker(seq, size):
"""
Yield successive chunks of size `size` from the list `seq`.
"""
for pos in range(0, len(seq), size):
yield seq[pos:pos + size]
def get_resources_with_tag(client, tag_key, tag_value, logger):
"""
Retrieve all AWS resources that have a specific tag key and value.
:param client: A boto3 client for the Resource Groups Tagging API.
:param tag_key: The tag key to filter resources (e.g., "Owner").
:param tag_value: The tag value to filter resources (e.g., "Cloud").
:param logger: The logger instance.
:return: List of resource ARNs.
"""
resource_arns = []
logger.info(f"Searching for resources with tag {tag_key}={tag_value}...")
paginator = client.get_paginator('get_resources')
try:
for page in paginator.paginate(
TagFilters=[{'Key': tag_key, 'Values': [tag_value]}]
):
mappings = page.get('ResourceTagMappingList', [])
logger.debug(f"Retrieved {len(mappings)} resources in current page.")
for resource in mappings:
arn = resource.get('ResourceARN')
if arn:
resource_arns.append(arn)
logger.debug(f"Found resource ARN: {arn}")
except ClientError as e:
logger.error(f"Error retrieving resources: {e}")
return resource_arns
def update_resource_tags(client, resource_arns, new_tag_key, new_tag_value, logger):
"""
Updates the tag for each resource ARN to the new key and value.
:param client: A boto3 client for the Resource Groups Tagging API.
:param resource_arns: List of resource ARNs to update.
:param new_tag_key: New tag key to apply.
:param new_tag_value: New tag value to apply.
:param logger: The logger instance.
"""
new_tags = {new_tag_key: new_tag_value}
if not resource_arns:
logger.info("No resources to update.")
return
logger.info(f"Updating tags for {len(resource_arns)} resources to {new_tag_key}={new_tag_value}...")
# AWS supports tagging up to 20 resources per call.
for arn_chunk in chunker(resource_arns, 20):
try:
response = client.tag_resources(
ResourceARNList=arn_chunk,
Tags=new_tags
)
logger.debug(f"Updated tags for resources: {arn_chunk}")
logger.debug(f"Response: {response}")
except ClientError as e:
logger.error(f"Error updating tags for resources {arn_chunk}: {e}")
def parse_args():
"""
Parses command-line arguments.
Required arguments:
--tag-filter: Tag filter in the format KEY=VALUE (e.g., Owner=NoCloud)
--tag-new: New tag in the format KEY=VALUE (e.g., Owner=Cloud)
Optional arguments:
--log-level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
"""
parser = argparse.ArgumentParser(description="Update AWS resource tags using boto3.")
parser.add_argument('--tag-filter', required=True,
help="Tag filter in the format KEY=VALUE (e.g., Owner=NoCloud)")
parser.add_argument('--tag-new', required=True,
help="New tag in the format KEY=VALUE (e.g., Owner=Cloud)")
parser.add_argument('--log-level', default='INFO',
choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
help="Set the logging level (default: INFO)")
return parser.parse_args()
def main():
args = parse_args()
logger = setup_logger(args.log_level)
logger.info("Starting AWS tag update script.")
# Parse the tag filter argument (e.g., Owner=NoCloud)
try:
filter_key, filter_value = args.tag_filter.split("=", 1)
except ValueError:
logger.error("Invalid --tag-filter format. Use KEY=VALUE (e.g., Owner=NoCloud).")
return
# Parse the new tag argument (e.g., Owner=Cloud)
try:
new_key, new_value = args.tag_new.split("=", 1)
except ValueError:
logger.error("Invalid --tag-new format. Use KEY=VALUE (e.g., Owner=Cloud).")
return
# Create a boto3 client for the Resource Groups Tagging API.
try:
client = boto3.client('resourcegroupstaggingapi')
except Exception as e:
logger.error(f"Failed to create boto3 client: {e}")
return
# Retrieve resources that match the specified tag filter.
resource_arns = get_resources_with_tag(client, filter_key, filter_value, logger)
if not resource_arns:
logger.info(f"No resources found with tag {filter_key}={filter_value}. Exiting.")
return
# Update the resources with the new tag.
update_resource_tags(client, resource_arns, new_key, new_value, logger)
logger.info("Tag update process completed.")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment