<filename>packages/validate_and_forward/lambda_handler.py<gh_stars>1-10
import json
import os
import traceback
from datetime import datetime
from uuid import uuid4

import boto3
from aws.ssm import get_ssm_params
from database import Jobs
from gp_file_parser.parser import parse_gp_extract_file_s3
from jobs.statuses import InputFolderType, InvalidErrorType, JobStatus, ParseStatus
from lr_logging import error, get_cloudlogbase_config, success
from lr_logging.exceptions import InvalidFilename, InvalidGPExtract, InvalidStructure
from mesh import AWSMESHMailbox, get_mesh_mailboxes
from pynamodb.exceptions import PutError, PynamoDBConnectionError, QueryError
from spine_aws_common.lambda_application import LambdaApplication


class ValidateAndForward(LambdaApplication):
    """
    Lambda:
        Responsible for creating a Job, tracking it DynamoDB and then sending a given input file off to DPS.
        If the input file is invalid, then reject
    Trigger:
        extracts_inbound inputs bucket
    """

    def __init__(self):
        super().__init__(additional_log_config=get_cloudlogbase_config())
        self.job_id = None
        self.s3 = boto3.client("s3")
        self.mesh_params = get_ssm_params(
            self.system_config["MESH_SSM_PREFIX"], self.system_config["AWS_REGION"]
        )

    def initialise(self):
        pass

    def start(self):
        try:
            self.job_id = str(uuid4())
            self.input_bucket = self.event["Records"][0]["s3"]["bucket"]["name"]
            self.input_file = self.event["Records"][0]["s3"]["object"]["key"]
        except KeyError as e:
            self.response = error(
                "validate_and_forward Lambda tried to access missing key",
                self.log_object.internal_id,
                error=traceback.format_exc(),
            )
            raise e

        self.create_initial_job()
        self.filename, file = self.get_file_contents()

        # get gp_file extract stats for job's table
        # Also handles rejection if file can't be parsed and validations fails
        try:
            practice_code, num_of_records = self.parse_gp_extract(
                self.input_bucket, self.input_file
            )

            self.log_object.write_log(
                "LRSDI02",
                log_row_dict={
                    "input_file": self.input_file,
                    "job_id": self.job_id,
                },
            )
        except (InvalidStructure, InvalidGPExtract, InvalidFilename) as exc:
            self.update_job_status(
                JobStatus.REJECTED.value, ParseStatus.PARSE_FAILED.value
            )

            message = json.dumps(self.process_invalid_message(exc))

            self.handle_rejection(InputFolderType.REJECTED, message)

            self.log_object.write_log(
                "LRSDW01",
                log_row_dict={
                    "input_file": self.input_file,
                    "job_id": self.job_id,
                    "reason": exc,
                },
            )

            self.response = success(
                message="Lambda application stopped gp extract input file rejected, rejected file handled",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
                reason=exc,
            )

            return self.response

        self.update_job_info(practice_code, num_of_records)
        self.send_mesh_file(self.filename, file)
        self.update_job_status(JobStatus.SENT_TO_DPS.value)

        self.response = success(
            message="validate_and_forward file sent",
            internal_id=self.log_object.internal_id,
            job=self.job_id,
            file=self.input_file,
        )

    def create_initial_job(self):

        """Creates an initial Job item in DynamoDb."""
        try:
            job_item = Jobs(
                self.job_id,
                PracticeCode=ParseStatus.NOT_PARSED.value,
                FileName=self.input_file,
                Timestamp=datetime.now(),
                StatusId=JobStatus.PENDING.value,
            )
            job_item.save()

            self.log_object.write_log(
                "LRSDI01",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                },
            )

        except (PynamoDBConnectionError, PutError) as e:

            self.log_object.write_log(
                "LRSDC01",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to create a job item before processing the gp_extract file",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )
            raise e

    def parse_gp_extract(self, input_bucket: str, input_file: str) -> tuple[str, int]:
        """Handler to process an uploaded S3 object containing a GP flat
            file extract

        Returns:
            extract file data: practice_code & num_of_records
        """
        self.upload_date = datetime.now()

        input_file_dict = parse_gp_extract_file_s3(
            self.s3,
            input_bucket,
            input_file,
            self.upload_date,
        )

        practice_code = input_file_dict["practice_code"]
        num_of_records = len(input_file_dict["records"])

        return practice_code, num_of_records

    def get_file_contents(self):
        filename = os.path.basename(str(self.input_file))
        file = (
            self.s3.get_object(
                Bucket=self.input_bucket,
                Key=self.input_file,
            )["Body"]
            .read()
            .decode("utf-8")
        )
        return filename, file

    def send_mesh_file(self, filename: str, file):

        listrec_mesh_id, dps_mesh_id = get_mesh_mailboxes(
            json.loads(self.mesh_params["mesh_mappings"]),
            self.mesh_params["listrec_dps_workflow"],
        )

        mesh = AWSMESHMailbox(listrec_mesh_id, self.log_object)
        mesh.send_message(dps_mesh_id, filename, file, overwrite=True)

        self.log_object.write_log(
            "LRSDI04",
            log_row_dict={
                "job_id": self.job_id,
                "input_file": self.input_file,
                "dps_mesh_id": dps_mesh_id,
            },
        )

    def update_job_info(self, practice_code: str, num_of_records: int):
        """Creates Job items in DynamoDb.

        Args:
            practice_code (str): GP practice code of GP extract.
            num_of_records (int): Number of records in GP extract.

        """
        try:
            job = Jobs.IdIndex.query(self.job_id)
            for j in job:
                j.PracticeCode = practice_code
                j.TotalRecords = num_of_records
                j.save()
            self.log_object.write_log(
                "LRSDI03",
                log_row_dict={
                    "upload_filename": self.input_file,
                    "job_id": self.job_id,
                },
            )
        except (PynamoDBConnectionError, PutError, QueryError) as e:

            self.log_object.write_log(
                "LRSDC02",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to update job before sending to DPS",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )
            raise e

    def update_job_status(self, status_code: str, practice_code=None):
        """Updates Job Status in DynamoDB"""
        try:
            job = Jobs.IdIndex.query(self.job_id)
            if practice_code is None:
                for j in job:
                    j.StatusId = status_code
                    j.save()
            else:
                for j in job:
                    j.StatusId = status_code
                    j.PracticeCode = practice_code
                    j.save()

            self.log_object.write_log(
                "LRSDI05",
                log_row_dict={
                    "upload_filename": self.input_file,
                    "job_id": self.job_id,
                    "status_code": status_code,
                },
            )
        except (PynamoDBConnectionError, PutError, QueryError) as e:

            self.log_object.write_log(
                "LRSDC03",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to update job after sending to DPS",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )

            raise e

    def cleanup_files(self, bucket: str, key: str):
        """Cleanup file that have already been processed e.g. sent DPS file from s3 or rejected"""
        try:
            self.s3.delete_object(Bucket=bucket, Key=key)

            self.log_object.write_log(
                "LRSDI03",
                log_row_dict={"key": key, "bucket": bucket},
            )

        except Exception as e:
            self.log_object.write_log(
                "LRSDC02",
                log_row_dict={"key": key, "bucket": bucket},
            )

            raise e

    def handle_rejection(self, prefix: InputFolderType, error_message: str = None):
        """Handles a rejected GP extract file,
            stores input file and log in output bucket

        Args:
            prefix (str): S3 folder prefix for where to place the handled file
            error_message (str): message to handle.
        """

        rejection_output_bucket = self.system_config["REJECTION_BUCKET"]
        key = f"{prefix.value}{self.filename}"

        self.s3.copy_object(
            Bucket=rejection_output_bucket,
            Key=key,
            CopySource={"Bucket": self.input_bucket, "Key": self.input_file},
        )

        if error_message:
            log_filename = f"{self.filename}-RejectedFile-{self.job_id}.json"
            log_key = f"{InputFolderType.REJECTED.value}logs/{log_filename}"

            self.s3.put_object(
                Body=error_message, Bucket=rejection_output_bucket, Key=log_key
            )

        self.cleanup_files(self.input_bucket, self.input_file)

    def process_invalid_message(self, exception: Exception) -> str:
        """Create a formatted error message string based on raised
            exception, used to store log files

        Args:
            exception (Exception): exception raised

        Returns:
            dict: dictionary of failed file information
        """

        rejection_log = {"file": self.input_file, "upload_date": str(self.upload_date)}

        if isinstance(exception, InvalidStructure):
            error = {
                "error_type": InvalidErrorType.STRUCTURE.value,
                "message": [exception.args[0]],
            }

        elif isinstance(exception, InvalidGPExtract):
            msg = exception.args[0]

            error = {
                "error_type": InvalidErrorType.RECORDS.value,
                "total_records": msg["total_records"],
                "total_invalid_records": len(msg["invalid_records"]),
                "message": msg["invalid_records"],
            }

        elif isinstance(exception, InvalidFilename):
            msg = exception.args[0]["message"]

            error = {"error_type": InvalidErrorType.FILENAME.value, "message": msg}

        rejection_log.update(error)

        return rejection_log

<filename>packages/validate_and_forward/lambda_handler.py<gh_stars>1-10
import json
import os
import traceback
from datetime import datetime
from uuid import uuid4

import boto3
from aws.ssm import get_ssm_params
from database import Jobs
from gp_file_parser.parser import parse_gp_extract_file_s3
from jobs.statuses import InputFolderType, InvalidErrorType, JobStatus, ParseStatus
from lr_logging import error, get_cloudlogbase_config, success
from lr_logging.exceptions import InvalidFilename, InvalidGPExtract, InvalidStructure
from mesh import AWSMESHMailbox, get_mesh_mailboxes
from pynamodb.exceptions import PutError, PynamoDBConnectionError, QueryError
from spine_aws_common.lambda_application import LambdaApplication


class ValidateAndForward(LambdaApplication):
    """
    Lambda:
        Responsible for creating a Job, tracking it DynamoDB and then sending a given input file off to DPS.
        If the input file is invalid, then reject
    Trigger:
        extracts_inbound inputs bucket
    """

    def __init__(self):
        super().__init__(additional_log_config=get_cloudlogbase_config())
        self.job_id = None
        self.s3 = boto3.client("s3")
        self.mesh_params = get_ssm_params(
            self.system_config["MESH_SSM_PREFIX"], self.system_config["AWS_REGION"]
        )

    def initialise(self):
        pass

    def start(self):
        try:
            self.job_id = str(uuid4())
            self.input_bucket = self.event["Records"][0]["s3"]["bucket"]["name"]
            self.input_file = self.event["Records"][0]["s3"]["object"]["key"]
        except KeyError as e:
            self.response = error(
                "validate_and_forward Lambda tried to access missing key",
                self.log_object.internal_id,
                error=traceback.format_exc(),
            )
            raise e

        self.create_initial_job()
        self.filename, file = self.get_file_contents()

        # get gp_file extract stats for job's table
        # Also handles rejection if file can't be parsed and validations fails
        try:
            practice_code, num_of_records = self.parse_gp_extract(
                self.input_bucket, self.input_file
            )

            self.log_object.write_log(
                "LRSDI02",
                log_row_dict={
                    "input_file": self.input_file,
                    "job_id": self.job_id,
                },
            )
        except (InvalidStructure, InvalidGPExtract, InvalidFilename) as exc:
            self.update_job_status(
                JobStatus.REJECTED.value, ParseStatus.PARSE_FAILED.value
            )

            message = json.dumps(self.process_invalid_message(exc))

            self.handle_rejection(InputFolderType.REJECTED, message)

            self.log_object.write_log(
                "LRSDW01",
                log_row_dict={
                    "input_file": self.input_file,
                    "job_id": self.job_id,
                    "reason": exc,
                },
            )

            self.response = success(
                message="Lambda application stopped gp extract input file rejected, rejected file handled",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
                reason=exc,
            )

            return self.response

        self.update_job_info(practice_code, num_of_records)
        self.send_mesh_file(self.filename, file)
        self.update_job_status(JobStatus.SENT_TO_DPS.value)

        self.response = success(
            message="validate_and_forward file sent",
            internal_id=self.log_object.internal_id,
            job=self.job_id,
            file=self.input_file,
        )

    def create_initial_job(self):

        """Creates an initial Job item in DynamoDb."""
        try:
            job_item = Jobs(
                self.job_id,
                PracticeCode=ParseStatus.NOT_PARSED.value,
                FileName=self.input_file,
                Timestamp=datetime.now(),
                StatusId=JobStatus.PENDING.value,
            )
            job_item.save()

            self.log_object.write_log(
                "LRSDI01",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                },
            )

        except (PynamoDBConnectionError, PutError) as e:

            self.log_object.write_log(
                "LRSDC01",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to create a job item before processing the gp_extract file",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )
            raise e

    def parse_gp_extract(self, input_bucket: str, input_file: str) -> tuple[str, int]:
        """Handler to process an uploaded S3 object containing a GP flat
            file extract

        Returns:
            extract file data: practice_code & num_of_records
        """
        self.upload_date = datetime.now()

        input_file_dict = parse_gp_extract_file_s3(
            self.s3,
            input_bucket,
            input_file,
            self.upload_date,
        )

        practice_code = input_file_dict["practice_code"]
        num_of_records = len(input_file_dict["records"])

        return practice_code, num_of_records

    def get_file_contents(self):
        filename = os.path.basename(str(self.input_file))
        file = (
            self.s3.get_object(
                Bucket=self.input_bucket,
                Key=self.input_file,
            )["Body"]
            .read()
            .decode("utf-8")
        )
        return filename, file

    def send_mesh_file(self, filename: str, file):

        listrec_mesh_id, dps_mesh_id = get_mesh_mailboxes(
            json.loads(self.mesh_params["mesh_mappings"]),
            self.mesh_params["listrec_dps_workflow"],
        )

        mesh = AWSMESHMailbox(listrec_mesh_id, self.log_object)
        mesh.send_message(dps_mesh_id, filename, file, overwrite=True)

        self.log_object.write_log(
            "LRSDI04",
            log_row_dict={
                "job_id": self.job_id,
                "input_file": self.input_file,
                "dps_mesh_id": dps_mesh_id,
            },
        )

    def update_job_info(self, practice_code: str, num_of_records: int):
        """Creates Job items in DynamoDb.

        Args:
            practice_code (str): GP practice code of GP extract.
            num_of_records (int): Number of records in GP extract.

        """
        try:
            job = Jobs.IdIndex.query(self.job_id)
            for j in job:
                j.PracticeCode = practice_code
                j.TotalRecords = num_of_records
                j.save()
            self.log_object.write_log(
                "LRSDI03",
                log_row_dict={
                    "upload_filename": self.input_file,
                    "job_id": self.job_id,
                },
            )
        except (PynamoDBConnectionError, PutError, QueryError) as e:

            self.log_object.write_log(
                "LRSDC02",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to update job before sending to DPS",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )
            raise e

    def update_job_status(self, status_code: str, practice_code=None):
        """Updates Job Status in DynamoDB"""
        try:
            job = Jobs.IdIndex.query(self.job_id)
            if practice_code is None:
                for j in job:
                    j.StatusId = status_code
                    j.save()
            else:
                for j in job:
                    j.StatusId = status_code
                    j.PracticeCode = practice_code
                    j.save()

            self.log_object.write_log(
                "LRSDI05",
                log_row_dict={
                    "upload_filename": self.input_file,
                    "job_id": self.job_id,
                    "status_code": status_code,
                },
            )
        except (PynamoDBConnectionError, PutError, QueryError) as e:

            self.log_object.write_log(
                "LRSDC03",
                log_row_dict={
                    "job_id": self.job_id,
                    "input_file": self.input_file,
                    "error": e,
                },
            )

            self.response = error(
                message="validate_and_forward failed to update job after sending to DPS",
                internal_id=self.log_object.internal_id,
                job_id=self.job_id,
                file=self.input_file,
            )

            raise e

    def cleanup_files(self, bucket: str, key: str):
        """Cleanup file that have already been processed e.g. sent DPS file from s3 or rejected"""
        try:
            self.s3.delete_object(Bucket=bucket, Key=key)

            self.log_object.write_log(
                "LRSDI03",
                log_row_dict={"key": key, "bucket": bucket},
            )

        except Exception as e:
            self.log_object.write_log(
                "LRSDC02",
                log_row_dict={"key": key, "bucket": bucket},
            )

            raise e

    def handle_rejection(self, prefix: InputFolderType, error_message: str = None):
        """Handles a rejected GP extract file,
            stores input file and log in output bucket

        Args:
            prefix (str): S3 folder prefix for where to place the handled file
            error_message (str): message to handle.
        """

        rejection_output_bucket = self.system_config["REJECTION_BUCKET"]
        key = f"{prefix.value}{self.filename}"

        self.s3.copy_object(
            Bucket=rejection_output_bucket,
            Key=key,
            CopySource={"Bucket": self.input_bucket, "Key": self.input_file},
        )

        if error_message:
            log_filename = f"{self.filename}-RejectedFile-{self.job_id}.json"
            log_key = f"{InputFolderType.REJECTED.value}logs/{log_filename}"

            self.s3.put_object(
                Body=error_message, Bucket=rejection_output_bucket, Key=log_key
            )

        self.cleanup_files(self.input_bucket, self.input_file)

    def process_invalid_message(self, exception: Exception) -> str:
        """Create a formatted error message string based on raised
            exception, used to store log files

        Args:
            exception (Exception): exception raised

        Returns:
            dict: dictionary of failed file information
        """

        rejection_log = {"file": self.input_file, "upload_date": str(self.upload_date)}

        if isinstance(exception, InvalidStructure):
            error = {
                "error_type": InvalidErrorType.STRUCTURE.value,
                "message": [exception.args[0]],
            }

        elif isinstance(exception, InvalidGPExtract):
            msg = exception.args[0]

            error = {
                "error_type": InvalidErrorType.RECORDS.value,
                "total_records": msg["total_records"],
                "total_invalid_records": len(msg["invalid_records"]),
                "message": msg["invalid_records"],
            }

        elif isinstance(exception, InvalidFilename):
            msg = exception.args[0]["message"]

            error = {"error_type": InvalidErrorType.FILENAME.value, "message": msg}

        rejection_log.update(error)

        return rejection_log

Lambda:         Responsible for creating a Job, tracking it DynamoDB and then sending a given input file off to DPS.         If the input file is invalid, then reject     Trigger:         extracts_inbound inputs bucket # get gp_file extract stats for job's table # Also handles rejection if file can't be parsed and validations fails Creates an initial Job item in DynamoDb. Handler to process an uploaded S3 object containing a GP flat             file extract          Returns:             extract file data: practice_code & num_of_records Creates Job items in DynamoDb.          Args:             practice_code (str): GP practice code of GP extract.             num_of_records (int): Number of records in GP extract. Updates Job Status in DynamoDB Cleanup file that have already been processed e.g. sent DPS file from s3 or rejected Handles a rejected GP extract file,             stores input file and log in output bucket          Args:             prefix (str): S3 folder prefix for where to place the handled file             error_message (str): message to handle. Create a formatted error message string based on raised             exception, used to store log files          Args:             exception (Exception): exception raised          Returns:             dict: dictionary of failed file information