VED-788 Refactor ack_backend

.github/dependabot.yml

@@ -7,7 +7,7 @@ version: 2
 updates:
   - package-ecosystem: "docker"
     directories:
-      - "/ack_backend"
+      - "/lambdas/ack_backend"
       - "/delta_backend"
       - "/filenameprocessor"
       - "/grafana/non-prod/docker"
@@ -49,7 +49,6 @@ updates:
   - package-ecosystem: "pip"
     directories:
       - "/"
-      - "/ack_backend"
       - "/backend"
       - "/batch_processor_filter"
       - "/delta_backend"
@@ -58,6 +57,7 @@ updates:
       - "/filenameprocessor"
       - "/mesh_processor"
       - "/recordprocessor"
+      - "/lambdas/ack_backend"
       - "/lambdas/redis_sync"
       - "/lambdas/id_sync"
       - "/lambdas/mns_subscription"

.github/workflows/quality-checks.yml

@@ -106,17 +106,6 @@ jobs:
           poetry run coverage run -m unittest discover -p "*batch*.py" || echo "recordforwarder tests failed" >> ../failed_tests.txt
           poetry run coverage xml -o ../recordforwarder-coverage.xml
 
-      - name: Run unittest with coverage-ack-lambda
-        working-directory: ack_backend
-        id: acklambda
-        env:
-          PYTHONPATH: ${{ github.workspace }}/ack_backend/src:${{ github.workspace }}/ack_backend/tests
-        continue-on-error: true
-        run: |
-          poetry install
-          poetry run coverage run -m unittest discover || echo "ack-lambda tests failed" >> ../failed_tests.txt
-          poetry run coverage xml -o ../ack-lambda-coverage.xml
-
       - name: Run unittest with coverage-delta
         working-directory: delta_backend
         id: delta
@@ -148,6 +137,17 @@ jobs:
           poetry run coverage run -m unittest discover || echo "mesh_processor tests failed" >> ../failed_tests.txt
           poetry run coverage xml -o ../mesh_processor-coverage.xml
 
+      - name: Run unittest with coverage-ack-lambda
+        working-directory: lambdas/ack_backend
+        id: acklambda
+        env:
+          PYTHONPATH: ${{ env.LAMBDA_PATH }}/ack_backend/src:${{ github.workspace }}/ack_backend/tests
+        continue-on-error: true
+        run: |
+            poetry install
+            poetry run coverage run --source=src -m unittest discover || echo "ack-lambda tests failed" >> ../../failed_tests.txt
+            poetry run coverage xml -o ../../ack-lambda-coverage.xml
+
       - name: Run unittest with coverage-mns-subscription
         working-directory: lambdas/mns_subscription
         id: mns_subscription

Makefile

@@ -1,6 +1,6 @@
 SHELL=/usr/bin/env bash -euo pipefail
 
-PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = ack_backend backend batch_processor_filter delta_backend filenameprocessor mesh_processor recordprocessor lambdas/redis_sync lambdas/id_sync lambdas/mns_subscription lambdas/shared
+PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = backend batch_processor_filter delta_backend filenameprocessor mesh_processor recordprocessor lambdas/ack_backend lambdas/redis_sync lambdas/id_sync lambdas/mns_subscription lambdas/shared
 PYTHON_PROJECT_DIRS = e2e e2e_batch $(PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS)
 
 #Installs dependencies using poetry.

immunisation-fhir-api.code-workspace

@@ -12,9 +12,6 @@
 		{
 			"path": "recordprocessor"
 		},
-		{
-			"path": "ack_backend"
-		},
 		{
 			"path": "delta_backend"
 		},
@@ -27,6 +24,9 @@
 		{
 			"path": "e2e_batch"
 		},
+		{
+			"path": "lambdas/ack_backend"
+		},
         {
             "path": "lambdas/redis_sync"
         },

lambdas/ack_backend/.vscode/settings.json.default

@@ -0,0 +1,18 @@
+{
+    "python.analysis.extraPaths": [
+        "./src"
+    ],
+    "python.testing.unittestArgs": [
+        "-v",
+        "-s",
+        "./",
+        "-p",
+        "test_*.py"
+    ],
+    "python.testing.pytestEnabled": false,
+    "python.testing.unittestEnabled": true,
+    "pylint.args": [
+        "--init-hook",
+        "import sys; sys.path.append('./src')"
+    ]
+}

ack_backend/Dockerfile → lambdas/ack_backend/Dockerfile

@@ -5,12 +5,31 @@ RUN mkdir -p /home/appuser && \
     echo 'appuser:x:1001:' >> /etc/group && \
     chown -R 1001:1001 /home/appuser && pip install "poetry~=2.1.4"
 
-COPY poetry.lock pyproject.toml README.md ./
+# Install Poetry dependencies
+# Copy ack_backend Poetry files
+COPY ./ack_backend/poetry.lock ./ack_backend/pyproject.toml ./
+
+# Install ack_backend dependencies
+WORKDIR /var/task
 RUN poetry config virtualenvs.create false && poetry install --no-interaction --no-ansi --no-root --only main
+
 # -----------------------------
 FROM base AS build
-COPY src .
+
+# Set working directory back to Lambda task root
+WORKDIR /var/task
+
+# Copy shared source code
+COPY ./shared/src/common ./common
+
+# Copy ack_backend source code
+COPY ./ack_backend/src .
+
+# Set correct permissions
 RUN chmod 644 $(find . -type f) && chmod 755 $(find . -type d)
+
 # Switch to the non-root user for running the container
 USER 1001:1001
+
+# Set the Lambda handler
 CMD ["ack_processor.lambda_handler"]

lambdas/ack_backend/Makefile

@@ -0,0 +1,22 @@
+TEST_ENV := @PYTHONPATH=src:tests:../shared/src
+
+build:
+	docker build -t ack-lambda-build .
+
+package: build
+	mkdir -p build
+	docker run --rm -v $(shell pwd)/build:/build ack-lambda-build
+
+test:
+	$(TEST_ENV) python -m unittest
+
+coverage-run:
+	$(TEST_ENV) coverage run --source=src -m unittest discover -v
+
+coverage-report:
+	$(TEST_ENV) coverage report -m
+
+coverage-html:
+	$(TEST_ENV) coverage html
+
+.PHONY: build package 

ack_backend/pyproject.toml → lambdas/ack_backend/pyproject.toml

@@ -5,7 +5,8 @@ description = ""
 authors = ["Your Name <[email protected]>"]
 readme = "README.md"
 packages = [
-    {include = "src"}
+    {include = "src"},
+    {include = "common", from = "../shared/src"}
 ]
 
 [tool.poetry.dependencies]

ack_backend/src/audit_table.py → lambdas/ack_backend/src/audit_table.py

@@ -1,7 +1,7 @@
 """Add the filename to the audit table and check for duplicates."""
 
-from clients import dynamodb_client, logger
-from errors import UnhandledAuditTableError
+from common.clients import dynamodb_client, logger
+from common.models.errors import UnhandledAuditTableError
 from constants import AUDIT_TABLE_NAME, FileStatus, AuditTableKeys
 
 

ack_backend/src/constants.py → lambdas/ack_backend/src/constants.py

@@ -4,14 +4,17 @@
 
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")
 
+
 def get_source_bucket_name() -> str:
     """Get the SOURCE_BUCKET_NAME environment from environment variables."""
     return os.getenv("SOURCE_BUCKET_NAME")
 
+
 def get_ack_bucket_name() -> str:
     """Get the ACK_BUCKET_NAME environment from environment variables."""
     return os.getenv("ACK_BUCKET_NAME")
 
+
 class FileStatus:
     """File status constants"""
 

ack_backend/src/convert_message_to_ack_row.py → lambdas/ack_backend/src/convert_message_to_ack_row.py

@@ -1,11 +1,10 @@
 """Functions for converting the incoming message body into a row of ack data"""
 
-from typing import Union
 from logging_decorators import convert_message_to_ack_row_logging_decorator
 from update_ack_file import create_ack_data
 
 
-def get_error_message_for_ack_file(message_diagnostics) -> Union[None, str]:
+def get_error_message_for_ack_file(message_diagnostics) -> None | str:
     """Determines and returns the error message to be displayed in the ack file"""
     if message_diagnostics is None:
         return None

ack_backend/src/logging_decorators.py → lambdas/ack_backend/src/logging_decorators.py

@@ -1,51 +1,21 @@
 """Decorators for logging and sending logs to Firehose"""
 
 import os
-import json
 import time
 from datetime import datetime
 from functools import wraps
-from clients import firehose_client, logger
-
+from common.log_decorator import generate_and_send_logs
 
+PREFIX = "ack_processor"
 STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME", "immunisation-fhir-api-internal-dev-splunk-firehose")
 
 
-def send_log_to_firehose(log_data: dict) -> None:
-    """Sends the log_message to Firehose"""
-    try:
-        record = {"Data": json.dumps({"event": log_data}).encode("utf-8")}
-        firehose_client.put_record(DeliveryStreamName=STREAM_NAME, Record=record)
-        logger.info("Log sent to Firehose")
-    except Exception as error:  # pylint:disable = broad-exception-caught
-        logger.exception("Error sending log to Firehose: %s", error)
-
-
-def generate_and_send_logs(
-    start_time: float,
-    base_log_data: dict,
-    additional_log_data: dict,
-    use_ms_precision: bool = False,
-    is_error_log: bool = False
-) -> None:
-    """Generates log data which includes the base_log_data, additional_log_data, and time taken (calculated using the
-    current time and given start_time) and sends them to Cloudwatch and Firehose."""
-    seconds_elapsed = time.time() - start_time
-    formatted_time_elapsed = f"{round(seconds_elapsed * 1000, 5)}ms" if use_ms_precision else \
-        f"{round(seconds_elapsed, 5)}s"
-
-    log_data = {**base_log_data, "time_taken": formatted_time_elapsed, **additional_log_data}
-    log_function = logger.error if is_error_log else logger.info
-    log_function(json.dumps(log_data))
-    send_log_to_firehose(log_data)
-
-
 def convert_message_to_ack_row_logging_decorator(func):
     """This decorator logs the information on the conversion of a single message to an ack data row"""
 
     @wraps(func)
     def wrapper(message, created_at_formatted_string):
-        base_log_data = {"function_name": f"ack_processor_{func.__name__}", "date_time": str(datetime.now())}
+        base_log_data = {"function_name": f"{PREFIX}_{func.__name__}", "date_time": str(datetime.now())}
         start_time = time.time()
 
         try:
@@ -66,13 +36,13 @@ def wrapper(message, created_at_formatted_string):
                 "operation_requested": message.get("operation_requested", "unknown"),
                 **process_diagnostics(diagnostics, file_key, message_id),
             }
-            generate_and_send_logs(start_time, base_log_data, additional_log_data, use_ms_precision=True)
+            generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data, use_ms_precision=True)
 
             return result
 
         except Exception as error:
             additional_log_data = {"status": "fail", "statusCode": 500, "diagnostics": str(error)}
-            generate_and_send_logs(start_time, base_log_data, additional_log_data, use_ms_precision=True,
+            generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data, use_ms_precision=True,
                                    is_error_log=True)
             raise
 
@@ -85,7 +55,7 @@ def upload_ack_file_logging_decorator(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
 
-        base_log_data = {"function_name": f"ack_processor_{func.__name__}", "date_time": str(datetime.now())}
+        base_log_data = {"function_name": f"{PREFIX}_{func.__name__}", "date_time": str(datetime.now())}
         start_time = time.time()
 
         # NB this doesn't require a try-catch block as the wrapped function never throws an exception
@@ -94,7 +64,7 @@ def wrapper(*args, **kwargs):
             message_for_logs = "Record processing complete"
             base_log_data.update(result)
             additional_log_data = {"status": "success", "statusCode": 200, "message": message_for_logs}
-            generate_and_send_logs(start_time, base_log_data, additional_log_data)
+            generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data)
         return result
 
     return wrapper
@@ -106,19 +76,19 @@ def ack_lambda_handler_logging_decorator(func):
     @wraps(func)
     def wrapper(event, context, *args, **kwargs):
 
-        base_log_data = {"function_name": f"ack_processor_{func.__name__}", "date_time": str(datetime.now())}
+        base_log_data = {"function_name": f"{PREFIX}_{func.__name__}", "date_time": str(datetime.now())}
         start_time = time.time()
 
         try:
             result = func(event, context, *args, **kwargs)
             message_for_logs = "Lambda function executed successfully!"
             additional_log_data = {"status": "success", "statusCode": 200, "message": message_for_logs}
-            generate_and_send_logs(start_time, base_log_data, additional_log_data)
+            generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data)
             return result
 
         except Exception as error:
             additional_log_data = {"status": "fail", "statusCode": 500, "diagnostics": str(error)}
-            generate_and_send_logs(start_time, base_log_data, additional_log_data, is_error_log=True)
+            generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data, is_error_log=True)
             raise
 
     return wrapper

ack_backend/src/update_ack_file.py → lambdas/ack_backend/src/update_ack_file.py

@@ -1,21 +1,21 @@
 """Functions for uploading the data to the ack file"""
 
-from io import StringIO, BytesIO
-from typing import Union, Optional
 from botocore.exceptions import ClientError
-from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name
+from io import StringIO, BytesIO
+from typing import Optional
 from audit_table import change_audit_table_status_to_processed
-from clients import get_s3_client, logger
-from utils_for_ack_lambda import get_row_count
+from common.clients import get_s3_client, logger
+from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name
 from logging_decorators import upload_ack_file_logging_decorator
+from utils_for_ack_lambda import get_row_count
 
 
 def create_ack_data(
     created_at_formatted_string: str,
     local_id: str,
     row_id: str,
     successful_api_response: bool,
-    diagnostics: Union[None, str] = None,
+    diagnostics: None | str = None,
     imms_id: str = None,
 ) -> dict:
     """Returns a dictionary containing the ack headers as keys, along with the relevant values."""