Created
February 12, 2025 07:55
-
-
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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