SFR-2291: Throw error if we fail to generate webpub (#422)

managers/rabbitmq.py

@@ -93,3 +93,11 @@ def acknowledgeMessageProcessed(self, deliveryTag):
             self.createRabbitConnection()
             self.createChannel()
             self.acknowledgeMessageProcessed(deliveryTag)
+
+    def reject_message(self, delivery_tag: str, requeue=False):
+        try:
+            self.channel.basic_reject(delivery_tag=delivery_tag, requeue=requeue)
+        except (ConnectionWrongStateError, StreamLostError):
+            self.createRabbitConnection()
+            self.createChannel()
+            self.reject_message(delivery_tag, requeue)      

processes/file/s3_files.py

@@ -8,6 +8,7 @@
 from ..core import CoreProcess
 from managers import S3Manager, RabbitMQManager
 from logger import createLog
+from utils import retry_request
 
 
 logger = createLog(__name__)
@@ -92,27 +93,33 @@ def process_files():
                 logger.info(f'Stored file in S3 for {file_url}')
             except Exception:
                 logger.exception(f'Failed to store file for file url: {file_url}')
+                rabbit_mq_manager.reject_message(delivery_tag=message_props.delivery_tag)
 
     @staticmethod
+    @retry_request()
     def get_file_contents(file_url: str):
-        file_url_response = requests.get(
-            file_url,
-            stream=True,
-            timeout=15,
-            headers={ 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5)' }
-        )
-
-        if file_url_response.status_code == 200:
+        try:
+            file_url_response = requests.get(
+                file_url,
+                stream=True,
+                timeout=15,
+                headers={ 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5)' }
+            )
+
+            file_url_response.raise_for_status()
+
             file_contents = bytes()
 
             for byte_chunk in file_url_response.iter_content(1024 * 250):
                 file_contents += byte_chunk
 
             return file_contents
-
-        raise Exception(f'Unable to fetch file from url: {file_url}')
+        except Exception as e:
+            logger.exception(f'Failed to get file contents from {file_url}')
+            raise e
 
     @staticmethod
+    @retry_request()
     def generate_webpub(file_root, bucket):
         webpub_conversion_url = os.environ['WEBPUB_CONVERSION_URL']
         s3_file_path = f'https://{bucket}.s3.amazonaws.com/{file_root}/META-INF/container.xml'
@@ -124,5 +131,6 @@ def generate_webpub(file_root, bucket):
             webpub_response.raise_for_status()
 
             return webpub_response.content
-        except Exception:
+        except Exception as e:
             logger.exception(f'Failed to generate webpub for {file_root}')
+            raise e

tests/unit/processes/file/test_s3_files_process.py

@@ -113,7 +113,7 @@ def test_get_file_contents_success(self, test_instance, mocker):
     def test_get_file_contents_error(self, test_instance, mocker):
         mock_get_request = mocker.patch.object(requests, 'get')
         mock_response = mocker.MagicMock()
-        mock_response.status_code = 500
+        mock_response.raise_for_status.side_effect = Exception
         mock_get_request.return_value = mock_response
 
         with pytest.raises(Exception):
@@ -139,9 +139,8 @@ def test_generate_webpub_error(self, mocker):
         mock_response.raise_for_status.side_effect = Exception
         mock_get_request.return_value = mock_response
 
-        test_webpub = S3Process.generate_webpub('testRoot', 'testBucket')
-
-        assert test_webpub == None
+        with pytest.raises(Exception):
+            S3Process.generate_webpub('testRoot', 'testBucket')
 
         mock_get_request.assert_called_once_with(
             'test_conversion_url/api/https%3A%2F%2FtestBucket.s3.amazonaws.com%2FtestRoot%2FMETA-INF%2Fcontainer.xml',

utils/__init__.py

@@ -0,0 +1 @@
+from .retry_request import retry_request

utils/retry_request.py

@@ -0,0 +1,23 @@
+import functools
+from requests import ConnectionError, Timeout
+import time
+
+
+def retry_request(max_retries: int=3, wait_seconds: int=60):
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            for attempt in range(max_retries):
+                try:
+                    return func(*args, **kwargs)
+                except (ConnectionError, Timeout) as e:
+                    if attempt < max_retries - 1:
+                        time.sleep(wait_seconds * (attempt + 1))
+                    else:
+                        raise e
+                except Exception as e:
+                    raise e
+
+        return wrapper
+
+    return decorator