[PRMP-1054] Creation of Report Orchestration Lambda

.github/workflows/base-lambdas-reusable-deploy-all.yml

@@ -796,3 +796,17 @@ jobs:
       lambda_layer_names: "core_lambda_layer"
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
+
+  deploy_report_orchestration_lambda:
+    name: Deploy Search Document Review
+    uses: ./.github/workflows/base-lambdas-reusable-deploy.yml
+    with:
+      environment: ${{ inputs.environment }}
+      python_version: ${{ inputs.python_version }}
+      build_branch: ${{ inputs.build_branch }}
+      sandbox: ${{ inputs.sandbox }}
+      lambda_handler_name: report_orchestration_handler
+      lambda_aws_name: reportOrchestration
+      lambda_layer_names: "core_lambda_layer,reports_lambda_layer"
+    secrets:
+      AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}

lambdas/handlers/report_orchestration_handler.py

@@ -0,0 +1,56 @@
+import os
+import tempfile
+from datetime import datetime, timedelta, timezone
+
+from repositories.reporting.reporting_dynamo_repository import ReportingDynamoRepository
+from services.reporting.excel_report_generator_service import ExcelReportGenerator
+from services.reporting.report_orchestration_service import ReportOrchestrationService
+from utils.audit_logging_setup import LoggingService
+from utils.decorators.ensure_env_var import ensure_environment_variables
+from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
+from utils.decorators.override_error_check import override_error_check
+from utils.decorators.set_audit_arg import set_request_context_for_logging
+
+logger = LoggingService(__name__)
+
+
+def calculate_reporting_window():
+    now = datetime.now(timezone.utc)
+    today_7am = now.replace(hour=7, minute=0, second=0, microsecond=0)
+
+    if now < today_7am:
+        today_7am -= timedelta(days=1)
+
+    yesterday_7am = today_7am - timedelta(days=1)
+
+    return (
+        int(yesterday_7am.timestamp()),
+        int(today_7am.timestamp()),
+    )
+
+@ensure_environment_variables(
+    names=["BULK_UPLOAD_REPORT_TABLE_NAME"]
+)
+@override_error_check
+@handle_lambda_exceptions
+@set_request_context_for_logging
+def lambda_handler(event, context):
+    logger.info("Report orchestration lambda invoked")
+    table_name = os.getenv("BULK_UPLOAD_REPORT_TABLE_NAME")
+
+    repository = ReportingDynamoRepository(table_name)
+    excel_generator = ExcelReportGenerator()
+
+    service = ReportOrchestrationService(
+        repository=repository,
+        excel_generator=excel_generator,
+    )
+
+    window_start, window_end = calculate_reporting_window()
+    tmp_dir = tempfile.mkdtemp()
+
+    service.process_reporting_window(
+        window_start_ts=window_start,
+        window_end_ts=window_end,
+        output_dir=tmp_dir,
+    )

lambdas/repositories/reporting/reporting_dynamo_repository.py

@@ -0,0 +1,35 @@
+from typing import Dict, List
+
+from boto3.dynamodb.conditions import Attr
+from services.base.dynamo_service import DynamoDBService
+from utils.audit_logging_setup import LoggingService
+
+logger = LoggingService(__name__)
+
+
+class ReportingDynamoRepository:
+    def __init__(self, table_name: str):
+        self.table_name = table_name
+        self.dynamo_service = DynamoDBService()
+
+    def get_records_for_time_window(
+        self,
+        start_timestamp: int,
+        end_timestamp: int,
+    ) -> List[Dict]:
+        logger.info(
+            f"Querying reporting table for window, "
+            f"table_name: {self.table_name}, "
+            f"start_timestamp: {start_timestamp}, "
+            f"end_timestamp: {end_timestamp}",
+        )
+
+        filter_expression = Attr("Timestamp").between(
+            start_timestamp,
+            end_timestamp,
+        )
+
+        return self.dynamo_service.scan_whole_table(
+            table_name=self.table_name,
+            filter_expression=filter_expression,
+        )

lambdas/services/reporting/excel_report_generator_service.py

@@ -0,0 +1,58 @@
+from datetime import datetime
+
+from openpyxl.workbook import Workbook
+from utils.audit_logging_setup import LoggingService
+
+logger = LoggingService(__name__)
+
+
+class ExcelReportGenerator:
+    def create_report_orchestration_xlsx(
+        self,
+        ods_code: str,
+        records: list[dict],
+        output_path: str,
+    ) -> str:
+        logger.info(
+            f"Creating Excel report for ODS code {ods_code} and records {len(records)}"
+        )
+        wb = Workbook()
+        ws = wb.active
+        ws.title = "Daily Upload Report"
+
+        # Report metadata
+        ws.append([f"ODS Code: {ods_code}"])
+        ws.append([f"Generated at (UTC): {datetime.utcnow().isoformat()}"])
+        ws.append([])
+
+        # Header row
+        ws.append(
+            [
+                "ID",
+                "Date",
+                "NHS Number",
+                "Uploader ODS",
+                "PDS ODS",
+                "Upload Status",
+                "Reason",
+                "File Path",
+            ]
+        )
+
+        for record in records:
+            ws.append(
+                [
+                    record.get("ID"),
+                    record.get("Date"),
+                    record.get("NhsNumber"),
+                    record.get("UploaderOdsCode"),
+                    record.get("PdsOdsCode"),
+                    record.get("UploadStatus"),
+                    record.get("Reason"),
+                    record.get("FilePath"),
+                ]
+            )
+
+        wb.save(output_path)
+        logger.info(f"Excel report written successfully for for ods code {ods_code}")
+        return output_path

lambdas/services/reporting/report_orchestration_service.py

@@ -0,0 +1,62 @@
+import tempfile
+from collections import defaultdict
+
+from utils.audit_logging_setup import LoggingService
+
+logger = LoggingService(__name__)
+
+
+class ReportOrchestrationService:
+    def __init__(
+        self,
+        repository,
+        excel_generator,
+    ):
+        self.repository = repository
+        self.excel_generator = excel_generator
+
+    def process_reporting_window(
+        self,
+        window_start_ts: int,
+        window_end_ts: int,
+        output_dir: str,
+    ):
+        records = self.repository.get_records_for_time_window(
+            window_start_ts,
+            window_end_ts,
+        )
+        if not records:
+            logger.info("No records found for reporting window")
+            return
+
+        records_by_ods = self.group_records_by_ods(records)
+
+        for ods_code, ods_records in records_by_ods.items():
+            logger.info(
+                f"Generating report for ODS ods_code = {ods_code} record_count = {len(ods_records)}"
+            )
+            self.generate_ods_report(ods_code, ods_records)
+        logger.info("Report orchestration completed")
+
+    @staticmethod
+    def group_records_by_ods(records: list[dict]) -> dict[str, list[dict]]:
+        grouped = defaultdict(list)
+        for record in records:
+            ods_code = record.get("UploaderOdsCode") or "UNKNOWN"
+            grouped[ods_code].append(record)
+        return grouped
+
+    def generate_ods_report(
+        self,
+        ods_code: str,
+        records: list[dict],
+    ):
+        with tempfile.NamedTemporaryFile(
+            suffix=f"_{ods_code}.xlsx",
+            delete=False,
+        ) as tmp:
+            self.excel_generator.create_report_orchestration_xlsx(
+                ods_code=ods_code,
+                records=records,
+                output_path=tmp.name,
+            )

lambdas/tests/unit/handlers/test_report_orchestration_handler.py

@@ -0,0 +1,74 @@
+from unittest import mock
+from unittest.mock import MagicMock
+import pytest
+from handlers.report_orchestration_handler import lambda_handler
+
+
+class FakeContext:
+    aws_request_id = "test-request-id"
+
+
+@pytest.fixture(autouse=True)
+def mock_env(monkeypatch):
+    monkeypatch.setenv("BULK_UPLOAD_REPORT_TABLE_NAME", "TestTable")
+
+
+@pytest.fixture
+def mock_logger(mocker):
+    return mocker.patch("handlers.report_orchestration_handler.logger", new=MagicMock())
+
+
+@pytest.fixture
+def mock_repo(mocker):
+    return mocker.patch(
+        "handlers.report_orchestration_handler.ReportingDynamoRepository",
+        autospec=True,
+    )
+
+
+@pytest.fixture
+def mock_excel_generator(mocker):
+    return mocker.patch(
+        "handlers.report_orchestration_handler.ExcelReportGenerator",
+        autospec=True,
+    )
+
+
+@pytest.fixture
+def mock_service(mocker):
+    return mocker.patch(
+        "handlers.report_orchestration_handler.ReportOrchestrationService",
+        autospec=True,
+    )
+
+
+@pytest.fixture
+def mock_window(mocker):
+    return mocker.patch(
+        "handlers.report_orchestration_handler.calculate_reporting_window",
+        return_value=(100, 200),
+    )
+
+
+def test_lambda_handler_calls_service(
+    mock_logger, mock_repo, mock_excel_generator, mock_service, mock_window
+):
+    lambda_handler(event={}, context=FakeContext())
+
+    mock_repo.assert_called_once_with("TestTable")
+    mock_excel_generator.assert_called_once_with()
+
+    mock_service.assert_called_once()
+    instance = mock_service.return_value
+    instance.process_reporting_window.assert_called_once_with(
+        window_start_ts=100,
+        window_end_ts=200,
+        output_dir=mock.ANY,
+    )
+
+    mock_logger.info.assert_any_call("Report orchestration lambda invoked")
+
+
+def test_lambda_handler_calls_window_function(mock_service, mock_window):
+    lambda_handler(event={}, context=FakeContext())
+    mock_window.assert_called_once()

lambdas/tests/unit/repositories/reporting/test_reporting_dynamo_repository.py

@@ -0,0 +1,41 @@
+from unittest.mock import MagicMock
+
+import pytest
+from repositories.reporting.reporting_dynamo_repository import ReportingDynamoRepository
+
+
+@pytest.fixture
+def mock_dynamo_service(mocker):
+    mock_service = mocker.patch(
+        "repositories.reporting.reporting_dynamo_repository.DynamoDBService"
+    )
+    instance = mock_service.return_value
+    instance.scan_whole_table = MagicMock()
+    return instance
+
+
+@pytest.fixture
+def reporting_repo(mock_dynamo_service):
+    return ReportingDynamoRepository(table_name="TestTable")
+
+
+def test_get_records_for_time_window_calls_scan(mock_dynamo_service, reporting_repo):
+    mock_dynamo_service.scan_whole_table.return_value = []
+
+    reporting_repo.get_records_for_time_window(100, 200)
+
+    mock_dynamo_service.scan_whole_table.assert_called_once()
+    assert "filter_expression" in mock_dynamo_service.scan_whole_table.call_args.kwargs
+
+
+def test_get_records_for_time_window_returns_empty_list(
+    mock_dynamo_service, reporting_repo
+):
+    start_ts = 0
+    end_ts = 50
+    mock_dynamo_service.scan_whole_table.return_value = []
+
+    result = reporting_repo.get_records_for_time_window(start_ts, end_ts)
+
+    assert result == []
+    mock_dynamo_service.scan_whole_table.assert_called_once()