[PRMT-71] Improve efficiency of LG download (#637)

* [PRMT-71] - use stream

* [PRMT-71] - removed commented code

* [PRMT-71] - reduced delay

* [PRMT-71] - fixed some comments

* [PRMT-71] - fixed tests

* [PRMT-71] - updated code to reflect comments

* [PRMT-153] - removed commented code

* [PRMT-71] - addressed comments

* [PRMT-71] - forced table column word to stay together

* [PRMT-71] - removed first variable

* [PRMT-71] - added service tests

* [PRMT-71] - removed comments

Author: PedroSoaresNHS

app/src/helpers/requests/getPresignedUrlForZip.test.ts

@@ -82,7 +82,7 @@ describe('getPresignedUrlForZip', () => {
         );
     });
 
-    it('wait for 10 secs before the 1st polling', async () => {
+    it('does not wait before the 1st polling', async () => {
         mockedAxios.post.mockResolvedValueOnce(mockJobIdResponse);
         mockedAxios.get.mockResolvedValueOnce(mockCompletedResponse);
 
@@ -92,11 +92,10 @@ describe('getPresignedUrlForZip', () => {
             baseUrl,
         });
 
-        expect(mockWaitForSeconds).toHaveBeenCalledTimes(1);
-        expect(mockWaitForSeconds).toHaveBeenCalledWith(0);
+        expect(mockWaitForSeconds).toHaveBeenCalledTimes(0);
     });
 
-    it('wait for 10 secs between every polling', async () => {
+    it('wait between every polling', async () => {
         mockedAxios.post.mockResolvedValueOnce(mockJobIdResponse);
         mockedAxios.get
             .mockResolvedValueOnce(mockPendingResponse)
@@ -110,13 +109,20 @@ describe('getPresignedUrlForZip', () => {
             baseUrl,
         });
 
-        expect(mockWaitForSeconds).toHaveBeenCalledTimes(4);
+        expect(mockWaitForSeconds).toHaveBeenCalledTimes(3);
         expect(mockWaitForSeconds).toHaveBeenCalledWith(0);
     });
 
-    it('throw an error if got pending status for 3 times', async () => {
+    it('throw an error if got pending status for 10 times', async () => {
         mockedAxios.post.mockResolvedValueOnce(mockJobIdResponse);
         mockedAxios.get
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
+            .mockResolvedValueOnce(mockPendingResponse)
             .mockResolvedValueOnce(mockPendingResponse)
             .mockResolvedValueOnce(mockPendingResponse)
             .mockResolvedValueOnce(mockPendingResponse);

app/src/helpers/requests/getPresignedUrlForZip.ts

@@ -7,7 +7,7 @@ import waitForSeconds from '../utils/waitForSeconds';
 import { DownloadManifestError } from '../../types/generic/errors';
 import { isRunningInCypress } from '../utils/isLocal';
 
-export const DELAY_BETWEEN_POLLING_IN_SECONDS = isRunningInCypress() ? 0 : 10;
+export const DELAY_BETWEEN_POLLING_IN_SECONDS = isRunningInCypress() ? 0 : 3;
 
 type Args = {
     nhsNumber: string;
@@ -33,9 +33,10 @@ const getPresignedUrlForZip = async (args: Args) => {
 
     const jobId = await requestJobId(args);
     let pendingCount = 0;
-
-    while (pendingCount < 3) {
-        await waitForSeconds(DELAY_BETWEEN_POLLING_IN_SECONDS);
+    while (pendingCount < 10) {
+        if (pendingCount > 0) {
+            await waitForSeconds(DELAY_BETWEEN_POLLING_IN_SECONDS);
+        }
         const pollingResponse = await pollForPresignedUrl({
             baseUrl,
             baseHeaders,

app/src/styles/App.scss

@@ -666,6 +666,11 @@ $hunit: '%';
         font-size: 1.2rem !important;
     }
 }
+@media (min-width: 380px) {
+    #available-files-table-title .table-column-header {
+        word-break: keep-all;
+    }
+}
 
 //Feedback
 #feedback-form {

lambdas/services/base/s3_service.py

@@ -176,3 +176,17 @@ def save_or_create_file(self, source_bucket: str, file_key: str, body: bytes):
         return self.client.put_object(
             Bucket=source_bucket, Key=file_key, Body=BytesIO(body)
         )
+
+    def get_object_stream(self, bucket: str, key: str):
+        response = self.client.get_object(Bucket=bucket, Key=key)
+        return response.get("Body")
+
+    def upload_file_obj(self, file_obj, s3_bucket_name: str, file_key: str):
+        try:
+            self.client.upload_fileobj(file_obj, s3_bucket_name, file_key)
+            logger.info(f"Uploaded file object to s3://{s3_bucket_name}/{file_key}")
+        except ClientError as e:
+            logger.error(
+                f"Failed to upload file object to s3://{s3_bucket_name}/{file_key} - {e}"
+            )
+            raise e

lambdas/services/generate_document_manifest_zip_service.py

@@ -1,7 +1,8 @@
 import os
 import shutil
-import tempfile
 import zipfile
+from contextlib import closing
+from io import BytesIO
 
 from botocore.exceptions import ClientError
 from enums.lambda_error import LambdaError
@@ -20,84 +21,77 @@ class DocumentManifestZipService:
     def __init__(self, zip_trace: DocumentManifestZipTrace):
         self.s3_service = S3Service()
         self.dynamo_service = DynamoDBService()
-        self.temp_output_dir = tempfile.mkdtemp()
-        self.temp_downloads_dir = tempfile.mkdtemp()
         self.zip_trace_object = zip_trace
         self.zip_output_bucket = os.environ["ZIPPED_STORE_BUCKET_NAME"]
         self.zip_trace_table = os.environ["ZIPPED_STORE_DYNAMODB_NAME"]
         self.zip_file_name = f"patient-record-{zip_trace.job_id}.zip"
-        self.zip_file_path = os.path.join(self.temp_output_dir, self.zip_file_name)
 
     def handle_zip_request(self):
-        self.update_processing_status()
-        self.download_documents_to_be_zipped()
-        self.zip_files()
-        self.upload_zip_file()
-        self.cleanup_temp_files()
+        self.update_status(TraceStatus.PROCESSING)
+        zip_buffer = self.stream_zip_documents()
+        self.upload_zip_file(zip_buffer)
         self.update_dynamo_with_fields({"job_status", "zip_file_location"})
 
-    def download_documents_to_be_zipped(self):
-        logger.info("Downloading documents to be zipped")
+    def stream_zip_documents(self) -> BytesIO:
+        logger.info("Streaming and zipping documents in-memory")
         documents = self.zip_trace_object.files_to_download
-        for document_location, document_name in documents.items():
-            self.download_file_from_s3(document_name, document_location)
 
-    def download_file_from_s3(self, document_name: str, document_location: str):
-        download_path = os.path.join(self.temp_downloads_dir, document_name)
-        file_bucket, file_key = self.get_file_bucket_and_key(document_location)
-        try:
-            self.s3_service.download_file(file_bucket, file_key, download_path)
-        except ClientError as e:
-            self.update_failed_status()
-            msg = f"{file_key} may reference missing file in s3 bucket: {file_bucket}"
-            logger.error(
-                f"{LambdaError.ZipServiceClientError.to_str()} {msg + str(e)}",
-                {"Result": "Failed to create document manifest"},
-            )
-            raise GenerateManifestZipException(
-                status_code=500, error=LambdaError.ZipServiceClientError
-            )
+        zip_buffer = BytesIO()
+
+        with zipfile.ZipFile(zip_buffer, "w", compression=zipfile.ZIP_DEFLATED) as zipf:
+            for document_location, document_name in documents.items():
+                file_bucket, file_key = self.get_file_bucket_and_key(document_location)
+                try:
+                    with closing(
+                        self.s3_service.get_object_stream(file_bucket, file_key)
+                    ) as s3_object_stream:
+                        with zipf.open(document_name, mode="w") as zip_member:
+                            shutil.copyfileobj(
+                                s3_object_stream, zip_member, length=64 * 1024
+                            )
+                except ClientError as e:
+                    self.update_status(TraceStatus.FAILED)
+                    msg = f"Failed to fetch S3 object {file_bucket}/{file_key}: {e}"
+                    logger.error(f"{LambdaError.ZipServiceClientError.to_str()} {msg}")
+                    raise GenerateManifestZipException(
+                        status_code=500, error=LambdaError.ZipServiceClientError
+                    )
+        zip_buffer.seek(0)
+        return zip_buffer
 
     def get_file_bucket_and_key(self, file_location: str):
         try:
             file_bucket, file_key = file_location.replace("s3://", "").split("/", 1)
             return file_bucket, file_key
         except ValueError:
-            self.update_failed_status()
+            self.update_status(TraceStatus.FAILED)
             raise InvalidDocumentReferenceException(
                 "Failed to parse bucket from file location string"
             )
 
-    def upload_zip_file(self):
+    def upload_zip_file(self, zip_buffer: BytesIO):
         logger.info("Uploading zip file to s3")
+        zip_file_key = f"{self.zip_file_name}"
         self.zip_trace_object.zip_file_location = (
-            f"s3://{self.zip_output_bucket}/{self.zip_file_name}"
+            f"s3://{self.zip_output_bucket}/{zip_file_key}"
         )
-
         try:
-            self.s3_service.upload_file(
-                file_name=self.zip_file_path,
-                s3_bucket_name=self.zip_output_bucket,
-                file_key=f"{self.zip_file_name}",
+            self.s3_service.upload_file_obj(
+                zip_buffer, self.zip_output_bucket, zip_file_key
             )
+            logger.info(
+                f"Successfully uploaded ZIP file to S3: s3://{self.zip_output_bucket}/{zip_file_key}"
+            )
+
         except ClientError as e:
-            self.update_failed_status()
+            self.update_status(TraceStatus.FAILED)
             logger.error(e, {"Result": "Failed to create document manifest"})
             raise GenerateManifestZipException(
                 status_code=500, error=LambdaError.ZipServiceClientError
             )
 
         self.zip_trace_object.job_status = TraceStatus.COMPLETED
 
-    def zip_files(self):
-        logger.info("Creating zip from files")
-        with zipfile.ZipFile(self.zip_file_path, "w", zipfile.ZIP_DEFLATED) as zipf:
-            for root, _, files in os.walk(self.temp_downloads_dir):
-                for file in files:
-                    file_path = os.path.join(root, file)
-                    arc_name = os.path.relpath(file_path, self.temp_downloads_dir)
-                    zipf.write(file_path, arc_name)
-
     def update_dynamo_with_fields(self, fields: set):
         logger.info("Writing zip trace to db")
         self.dynamo_service.update_item(
@@ -108,14 +102,6 @@ def update_dynamo_with_fields(self, fields: set):
             ),
         )
 
-    def update_processing_status(self):
-        self.zip_trace_object.job_status = TraceStatus.PROCESSING
+    def update_status(self, job_status: TraceStatus):
+        self.zip_trace_object.job_status = job_status
         self.update_dynamo_with_fields({"job_status"})
-
-    def update_failed_status(self):
-        self.zip_trace_object.job_status = TraceStatus.FAILED
-        self.update_dynamo_with_fields({"job_status"})
-
-    def cleanup_temp_files(self):
-        shutil.rmtree(self.temp_downloads_dir)
-        shutil.rmtree(self.temp_output_dir)

lambdas/services/pds_api_service.py

@@ -47,7 +47,7 @@ def pds_request(self, nhs_number: str, retry_on_expired: bool):
             pds_response = self.session.get(
                 url=url_endpoint, headers=authorization_header
             )
-            logger.info("PDS Call Completed",{ "Event": "NDR-TR1" })
+            logger.info("PDS Call Completed", {"Event": "NDR-TR1"})
 
             if pds_response.status_code == 401 and retry_on_expired:
                 return self.pds_request(nhs_number, retry_on_expired=False)

lambdas/tests/unit/services/base/test_s3_service.py

@@ -410,7 +410,7 @@ def test_save_or_create_file(mock_service, mock_client):
     assert kwargs["Bucket"] == MOCK_BUCKET
     assert kwargs["Key"] == TEST_FILE_NAME
     assert kwargs["Body"].read() == body
-    
+
 
 def test_returns_binary_file_content_when_file_exists(
     mock_service, mock_client, mocker
@@ -431,3 +431,44 @@ def test_raises_exception_when_file_does_not_exist(mock_service, mock_client):
     with pytest.raises(ClientError):
         mock_service.get_binary_file("test-bucket", "nonexistent-key")
 
+
+def test_get_object_stream_returns_body_stream(mock_service, mock_client, mocker):
+    mock_stream = mocker.Mock(name="MockS3BodyStream")
+    mock_client.get_object.return_value = {"Body": mock_stream}
+
+    result = mock_service.get_object_stream(bucket=MOCK_BUCKET, key=TEST_FILE_KEY)
+
+    assert result == mock_stream
+    mock_client.get_object.assert_called_once_with(
+        Bucket=MOCK_BUCKET, Key=TEST_FILE_KEY
+    )
+
+
+def test_upload_file_obj_success(mock_service, mock_client, mocker):
+    mock_file_obj = mocker.Mock(name="MockFileObj")
+
+    mock_service.upload_file_obj(mock_file_obj, MOCK_BUCKET, TEST_FILE_KEY)
+
+    mock_client.upload_fileobj.assert_called_once_with(
+        mock_file_obj, MOCK_BUCKET, TEST_FILE_KEY
+    )
+
+
+def test_upload_file_obj_raises_client_error(mock_service, mock_client, mocker):
+    mock_file_obj = mocker.Mock(name="MockFileObj")
+    mock_client.upload_fileobj.side_effect = MOCK_CLIENT_ERROR
+
+    with pytest.raises(ClientError) as except_info:
+        mock_service.upload_file_obj(mock_file_obj, MOCK_BUCKET, TEST_FILE_KEY)
+
+    mock_client.upload_fileobj.assert_called_once_with(
+        mock_file_obj, MOCK_BUCKET, TEST_FILE_KEY
+    )
+    assert (
+        except_info.value.response["Error"]["Code"]
+        == MOCK_CLIENT_ERROR.response["Error"]["Code"]
+    )
+    assert (
+        except_info.value.response["Error"]["Message"]
+        == MOCK_CLIENT_ERROR.response["Error"]["Message"]
+    )