adding error handling with IAM policy deletes (race condition found)

Author: cyphronix

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/sra_iam.py

@@ -165,7 +165,15 @@ def detach_policy(self, role_name: str, policy_arn: str) -> EmptyResponseMetadat
             Empty response metadata
         """
         self.LOGGER.info("Detaching policy from %s.", role_name)
-        return self.IAM_CLIENT.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn)
+        try:
+            response = self.IAM_CLIENT.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn)
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "NoSuchEntity":
+                self.LOGGER.info(f"Policy '{policy_arn}' is not attached to role '{role_name}'.")
+            else:
+                self.LOGGER.error(f"Error detaching policy '{policy_arn}' from role '{role_name}': {error}")
+                raise ValueError(f"Error detaching policy '{policy_arn}' from role '{role_name}': {error}") from None
+        return response
 
     def delete_policy(self, policy_arn: str) -> EmptyResponseMetadataTypeDef:
         """Delete IAM Policy.
@@ -184,10 +192,25 @@ def delete_policy(self, policy_arn: str) -> EmptyResponseMetadataTypeDef:
             for version in page["Versions"]:
                 if not version["IsDefaultVersion"]:
                     self.LOGGER.info(f"Deleting policy version {version['VersionId']}")
-                    self.IAM_CLIENT.delete_policy_version(PolicyArn=policy_arn, VersionId=version["VersionId"])
-                    sleep(1)
-                    self.LOGGER.info("Policy version deleted.")
-        return self.IAM_CLIENT.delete_policy(PolicyArn=policy_arn)
+                    try:
+                        self.IAM_CLIENT.delete_policy_version(PolicyArn=policy_arn, VersionId=version["VersionId"])
+                        sleep(1)
+                        self.LOGGER.info("Policy version deleted.")
+                    except ClientError as error:
+                        if error.response["Error"]["Code"] == "NoSuchEntity":
+                            self.LOGGER.info(f"Policy version {version['VersionId']} not found.")
+                        else:
+                            self.LOGGER.error(f"Error deleting policy version {version['VersionId']}: {error}")
+                            raise ValueError(f"Error deleting policy version {version['VersionId']}: {error}") from None
+        try:
+            response = self.IAM_CLIENT.delete_policy(PolicyArn=policy_arn)
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "NoSuchEntity":
+                self.LOGGER.info(f"Policy {policy_arn} not found.")
+            else:
+                self.LOGGER.error(f"Error deleting policy {policy_arn}: {error}")
+                raise ValueError(f"Error deleting policy {policy_arn}: {error}") from None
+        return response
 
     def delete_role(self, role_name: str) -> EmptyResponseMetadataTypeDef:
         """Delete IAM role.