|
| 1 | +import sys |
| 2 | + |
| 3 | +import boto3 |
| 4 | +from botocore.exceptions import ClientError |
| 5 | + |
| 6 | + |
| 7 | +class CleanupTerraformStates: |
| 8 | + def __init__(self): |
| 9 | + self.env_folder = "env:/" |
| 10 | + self.client = boto3.client("s3") |
| 11 | + self.paginator = self.client.get_paginator('list_objects_v2') |
| 12 | + |
| 13 | + def get_terraform_bucket(self) -> str: |
| 14 | + response = self.client.list_buckets() |
| 15 | + buckets = response.get("Buckets") |
| 16 | + for bucket in buckets: |
| 17 | + if "ndr-dev-terraform-state" in bucket.get("Name"): |
| 18 | + bucket_name = bucket.get("Name") |
| 19 | + if not bucket_name: |
| 20 | + break |
| 21 | + return bucket_name |
| 22 | + print("Failed to find terraform bucket") |
| 23 | + sys.exit(1) |
| 24 | + |
| 25 | + def remove_folder_objects(self, bucket_name:str, folder_prefix: str): |
| 26 | + print(f"Deleting all objects under: {folder_prefix}") |
| 27 | + pages = self.paginator.paginate(Bucket=bucket_name, Prefix=folder_prefix) |
| 28 | + |
| 29 | + objects_to_delete = [] |
| 30 | + for page in pages: |
| 31 | + for obj in page.get('Contents', []): |
| 32 | + objects_to_delete.append({'Key': obj['Key']}) |
| 33 | + |
| 34 | + if objects_to_delete: |
| 35 | + self.client.delete_objects(Bucket=bucket_name, Delete={'Objects': objects_to_delete}) |
| 36 | + |
| 37 | + |
| 38 | + def empty_folder_check(self, bucket_name:str, folder_prefix: str): |
| 39 | + try: |
| 40 | + self.client.head_object(Bucket=bucket_name, Key=folder_prefix) |
| 41 | + self.client.delete_object(Bucket=bucket_name, Key=folder_prefix) |
| 42 | + print(f"Deleted empty folder for: {folder_prefix}") |
| 43 | + except ClientError as e: |
| 44 | + if e.response['Error']['Code'] != "404": |
| 45 | + print(f"No empty folder found to remove for: {folder_prefix}") |
| 46 | + |
| 47 | + def main(self, sandbox: str): |
| 48 | + tf_bucket = self.get_terraform_bucket() |
| 49 | + pages = self.paginator.paginate(Bucket=tf_bucket, Prefix=self.env_folder) |
| 50 | + |
| 51 | + for page in pages: |
| 52 | + for obj in page.get('Contents', []): |
| 53 | + key = obj['Key'] |
| 54 | + parent_folder = key[len(self.env_folder):].split("/", 1)[0] |
| 55 | + if parent_folder == sandbox: |
| 56 | + print(parent_folder) |
| 57 | + folder_prefix = f"{self.env_folder}{parent_folder}/" |
| 58 | + self.remove_folder_objects(bucket_name=tf_bucket, folder_prefix=folder_prefix) |
| 59 | + self.empty_folder_check(bucket_name=tf_bucket, folder_prefix=folder_prefix) |
| 60 | + |
| 61 | +if __name__ == '__main__': |
| 62 | + sandbox = sys.argv[1] |
| 63 | + exclude_list = ['ndr-dev'] |
| 64 | + |
| 65 | + if sandbox in exclude_list: |
| 66 | + print("Cleanup failed. Cannot delete protected environment") |
| 67 | + sys.exit(1) |
| 68 | + |
| 69 | + print(f"Attempting to cleanup the terraform states for: {sandbox}") |
| 70 | + CleanupTerraformStates().main(sandbox=sandbox) |
0 commit comments