Method handle_error

scripts/agents/devops/src/aws/base.py:197–273

Handle common AWS errors and convert to appropriate exceptions. Args: error: The exception that was raised. operation: The name of the operation that failed. Raises: ResourceNotFoundError: If the resource was not foun

(self, error: Exception, operation: str)

            raise AWSServiceError(f"Failed to verify AWS access: {str(e)}")
    
    def handle_error(self, error: Exception, operation: str) -> None:
        """
        Handle common AWS errors and convert to appropriate exceptions.
        
        Args:
            error: The exception that was raised.
            operation: The name of the operation that failed.
            
        Raises:
            ResourceNotFoundError: If the resource was not found.
            PermissionDeniedError: If permissions are insufficient.
            ValidationError: If input validation failed.
            RateLimitError: If rate limits were exceeded.
            ResourceLimitError: If resource limits were exceeded.
            AWSServiceError: For other AWS-related errors.
        """
        if isinstance(error, ClientError):
            error_code = error.response.get('Error', {}).get('Code', '')
            
            if error_code in ['ResourceNotFoundException', 'NoSuchEntity', 'NoSuchBucket', 
                             'NotFound', 'InvalidInstanceID.NotFound', 'InvalidGroup.NotFound',
                             'InvalidSecurityGroupID.NotFound', 'InvalidKeyPair.NotFound',
                             'InvalidKeyPair.Duplicate', 'InvalidVpcID.NotFound']:
                # Extract resource type and ID from error message if possible
                resource_type = None
                resource_id = None
                
                error_msg = error.response.get('Error', {}).get('Message', '')
                if 'instance' in error_code.lower() or 'instance' in error_msg.lower():
                    resource_type = 'instance'
                elif 'security group' in error_msg.lower():
                    resource_type = 'security group'
                elif 'key pair' in error_msg.lower():
                    resource_type = 'key pair'
                elif 'vpc' in error_code.lower() or 'vpc' in error_msg.lower():
                    resource_type = 'VPC'
                
                # Try to extract resource ID from error message
                import re
                id_match = re.search(r"'([a-zA-Z0-9-]+)'", error_msg)
                if id_match:
                    resource_id = id_match.group(1)
                
                raise ResourceNotFoundError(f"{operation} failed: Resource not found", resource_type, resource_id)
            
            elif error_code in ['AccessDenied', 'UnauthorizedOperation']:
                raise PermissionDeniedError(f"{operation} failed: Permission denied - {error.response.get('Error', {}).get('Message', '')}")
            
            elif error_code in ['ValidationError', 'InvalidParameterValue', 'MalformedQueryString']:
                raise ValidationError(f"{operation} failed: Invalid parameters - {error.response.get('Error', {}).get('Message', '')}")
            
            elif error_code in ['Throttling', 'ThrottlingException', 'RequestLimitExceeded']:
                # Try to extract wait time from error message
                import re
                wait_time_match = re.search(r"try again in (\d+) seconds", str(error))
                wait_time = int(wait_time_match.group(1)) if wait_time_match else None
                raise RateLimitError(f"{operation} failed: Rate limit exceeded - {error.response.get('Error', {}).get('Message', '')}", wait_time)