VED-804 Move record processor to Python Lambdas directory (#935)

Author: dlzhry2nhs

.github/dependabot.yml

@@ -7,12 +7,13 @@ version: 2
 updates:
   - package-ecosystem: "docker"
     directories:
-      - "/lambdas/ack_backend"
-      - "/lambdas/delta_backend"
+      - "/batch_processor_filter"
       - "/filenameprocessor"
       - "/infrastructure/grafana/non-prod/docker"
+      - "/lambdas/ack_backend"
+      - "/lambdas/delta_backend"
+      - "/lambdas/recordprocessor"
       - "/mesh_processor"
-      - "/recordprocessor"
       - "/sandbox"
     schedule:
       interval: "daily"
@@ -54,7 +55,7 @@ updates:
       - "/tests/e2e_batch"
       - "/filenameprocessor"
       - "/mesh_processor"
-      - "/recordprocessor"
+      - "/lambdas/recordprocessor"
       - "/lambdas/ack_backend"
       - "/lambdas/delta_backend"
       - "/lambdas/redis_sync"

.github/workflows/quality-checks.yml

@@ -118,15 +118,15 @@ jobs:
           poetry run coverage xml -o ../batchprocessorfilter-coverage.xml
 
       - name: Run unittest with recordprocessor-coverage
-        working-directory: recordprocessor
+        working-directory: lambdas/recordprocessor
         id: recordprocessor
         env:
-          PYTHONPATH: ${{ github.workspace }}/recordprocessor/src:${{ github.workspace }}/recordprocessor/tests
+          PYTHONPATH: ${{ env.LAMBDA_PATH }}/recordprocessor/src:${{ env.LAMBDA_PATH }}/recordprocessor/tests:${{ env.SHARED_PATH }}/src
         continue-on-error: true
         run: |
           poetry install
-          poetry run coverage run -m unittest discover || echo "recordprocessor tests failed" >> ../failed_tests.txt
-          poetry run coverage xml -o ../recordprocessor-coverage.xml
+          poetry run coverage run --source=src -m unittest discover || echo "recordprocessor tests failed" >> ../../failed_tests.txt
+          poetry run coverage xml -o ../../recordprocessor-coverage.xml
 
         # This step is redundant - all of these tests will be run in the backend step below
       - name: Run unittest with recordforwarder-coverage

Makefile

@@ -1,6 +1,6 @@
 SHELL=/usr/bin/env bash -euo pipefail
 
-PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = backend batch_processor_filter filenameprocessor mesh_processor recordprocessor lambdas/ack_backend lambdas/delta_backend lambdas/redis_sync lambdas/id_sync lambdas/mns_subscription lambdas/shared
+PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = backend batch_processor_filter filenameprocessor mesh_processor lambdas/recordprocessor lambdas/ack_backend lambdas/delta_backend lambdas/redis_sync lambdas/id_sync lambdas/mns_subscription lambdas/shared
 PYTHON_PROJECT_DIRS = tests/e2e tests/e2e_batch quality_checks $(PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS)
 
 .PHONY: install lint format format-check clean publish build-proxy release initialise-all-python-venvs update-all-python-dependencies run-all-python-unit-tests build-all-docker-images

immunisation-fhir-api.code-workspace

@@ -10,7 +10,7 @@
             "path": "filenameprocessor",
         },
         {
-            "path": "recordprocessor",
+            "path": "lambdas/recordprocessor",
         },
         {
             "path": "mesh_processor",

infrastructure/instance/ecs_batch_processor_config.tf

@@ -5,7 +5,7 @@ resource "aws_ecs_cluster" "ecs_cluster" {
 
 # Locals for Lambda processing paths and hash
 locals {
-  processing_lambda_dir     = abspath("${path.root}/../../recordprocessor")
+  processing_lambda_dir     = abspath("${path.root}/../../lambdas/recordprocessor")
   processing_path_include   = ["**"]
   processing_path_exclude   = ["**/__pycache__/**"]
   processing_files_include  = setunion([for f in local.processing_path_include : fileset(local.processing_lambda_dir, f)]...)
@@ -29,8 +29,9 @@ module "processing_docker_image" {
   source  = "terraform-aws-modules/lambda/aws//modules/docker-build"
   version = "8.1.0"
 
-  create_ecr_repo = false
-  ecr_repo        = aws_ecr_repository.processing_repository.name
+  create_ecr_repo  = false
+  docker_file_path = "./recordprocessor/Dockerfile"
+  ecr_repo         = aws_ecr_repository.processing_repository.name
   ecr_repo_lifecycle_policy = jsonencode({
     "rules" : [
       {
@@ -50,9 +51,10 @@ module "processing_docker_image" {
 
   platform      = "linux/amd64"
   use_image_tag = false
-  source_path   = local.processing_lambda_dir
+  source_path   = abspath("${path.root}/../../lambdas")
   triggers = {
-    dir_sha = local.processing_lambda_dir_sha
+    dir_sha        = local.processing_lambda_dir_sha
+    shared_dir_sha = local.shared_dir_sha
   }
 }
 

recordprocessor/Dockerfile lambdas/recordprocessor/Dockerfilerecordprocessor/Dockerfile renamed to lambdas/recordprocessor/Dockerfile

@@ -6,23 +6,25 @@ RUN mkdir -p /home/appuser && \
     echo 'appuser:x:1001:' >> /etc/group && \
     chown -R 1001:1001 /home/appuser && pip install "poetry~=2.1.4"
 
-# Install Poetry as root
-COPY poetry.lock pyproject.toml README.md ./
-RUN poetry config virtualenvs.create false && poetry install --no-interaction --no-ansi --no-root --only main
+# Copy recordprocessor Poetry files
+COPY ./recordprocessor/poetry.lock ./recordprocessor/pyproject.toml ./
 
-# -----------------------------
-FROM base AS test
-COPY src src
-COPY tests tests
-RUN poetry install --no-interaction --no-ansi --no-root && \
-    pytest --disable-warnings tests
+# Install 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 .
+# Copy shared source code
+COPY ./shared/src/common ./common
+
+# Copy recordprocessor source code
+COPY ./recordprocessor/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
-CMD ["python", "batch_processor.py"]
 ENTRYPOINT ["python", "batch_processor.py"]

recordprocessor/Makefile lambdas/recordprocessor/Makefilerecordprocessor/Makefile renamed to lambdas/recordprocessor/Makefile

@@ -1,3 +1,5 @@
+TEST_ENV := @PYTHONPATH=src:tests:../shared/src
+
 build:
 	docker build -t processor-lambda-build .
 
@@ -6,15 +8,15 @@ package:build
 	docker run --rm -v $(shell pwd)/build:/build processor-lambda-build
 
 test:
-	@PYTHONPATH=src:tests python -m unittest
+	$(TEST_ENV) python -m unittest
 
 coverage-run:
-	@PYTHONPATH=src:tests coverage run -m unittest discover
+	$(TEST_ENV) coverage run -m unittest discover
 
 coverage-report:
-	coverage report -m
+	$(TEST_ENV) coverage report -m
 
 coverage-html:
-	coverage html
+	$(TEST_ENV) coverage html
 
 .PHONY: build package

lambdas/recordprocessor/README.md

@@ -0,0 +1,28 @@
+# record-processor project
+
+## Overview
+
+The Record Processor component is a core part of the batch processing system. The functionality is as follows:
+
+- receive a batch job event from EventBridge pipes. The job will contain the filename, so that the batch .csv/.dat file
+  can be retrieved, along with other metadata such as the supplier name and their permissions.
+- perform high-level validation e.g. marking a job as processed if the file is empty or failed if the supplier does not
+  have permissions to interact with the requested vaccination type.
+- reads the batch source file and processes the content row by row.
+- for each of the rows, it validates whether the supplier can perform the given operation i.e. CREATE, UPDATE or DELETE
+  and also checks for critical information such as the unique ID.
+- finally, if such checks pass, the flat CSV structure will be converted to an R4 Immunization FHIR JSON payload. The
+  mapping is performed in `utils_for_fhir_conversion.py`.
+- the content is sent to Kinesis for further downstream processing where the requested operations will be performed on
+  the IEDS table.
+
+For more context, refer to the [Architecture Overview](https://nhsd-confluence.digital.nhs.uk/spaces/Vacc/pages/1035417049/Immunisation+FHIR+API+-+Solution+Architecture) in Confluence.
+
+Finally, it is worth noting that this package is **not** deployed as a Lambda function. As the file processing can take
+some time for particularly large extracts, this is run in AWS ECS.
+
+## Set up
+
+Simply follow the instructions from the root README on _Setting up a virtual environment with poetry_.
+As is the case for developing with any of the Python projects, it is easiest to create a `.env` file, a `.envrc` file,
+and then install the poetry dependencies using `poetry install --no-root`.

recordprocessor/poetry.lock lambdas/recordprocessor/poetry.lockrecordprocessor/poetry.lock renamed to lambdas/recordprocessor/poetry.lock

@@ -4,7 +4,10 @@ version     = "0.1.0"
 description = ""
 authors     = ["Your Name <you@example.com>"]
 readme      = "README.md"
-packages    = [{include = "src"}]
+packages = [
+    {include = "src"},
+    {include = "common", from = "../shared/src"}
+]
 
 [tool.poetry.dependencies]
 python                  = "~3.11"