VED-470: Handle receiving multipart content from MESH.

Author: mfjarvis

mesh_processor/poetry.lock

@@ -10,7 +10,7 @@ packages    = [{include = "src"}]
 python = "~3.10"
 boto3 = "~1.38.42"
 mypy-boto3-dynamodb = "^1.38.4"
-moto = "~5.1.6"
+moto = {extras = ["s3"], version = "^5.1.8"}
 coverage = "^7.9.1"
 
 [build-system]

mesh_processor/pyproject.toml

@@ -1,44 +1,84 @@
+import logging
+
 import boto3
 import os
 
 
-def lambda_handler(event, context):
-    s3 = boto3.client('s3')
+DESTINATION_BUCKET_NAME = os.getenv("DESTINATION_BUCKET_NAME")
 
-    #  Destination bucket name
-    destination_bucket = os.getenv("Destination_BUCKET_NAME")
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger()
 
-    for record in event["Records"]:
-        bucket_name = record["s3"]["bucket"]["name"]
-        file_key = record["s3"]["object"]["key"]
-        copy_source = {
-            'Bucket': record["s3"]["bucket"]["name"],
-            'Key': record["s3"]["object"]["key"]
-        }
+s3_client = boto3.client('s3')
 
-    # Read the .dat file from S3
-    dat_obj = s3.get_object(Bucket=bucket_name, Key=file_key)
 
-    # Update the filename from Metadata
-    file_name = ensure_dat_extension(dat_obj['Metadata'].get('mex-filename', None))
+def parse_headers(headers_str: str):
+    headers = dict(
+        header_str.split(":", 1)
+        for header_str in headers_str.split("\r\n")
+        if ":" in header_str
+    )
+    return {k.strip(): v.strip() for k, v in headers.items()}
 
-    s3.copy_object(CopySource=copy_source, Bucket=destination_bucket, Key=file_name)
 
-    return {
-        'statusCode': 200,
-        'body': 'Files converted and uploaded successfully!'
-    }
+def parse_header_value(header_value: str):
+    main_value, *params = header_value.split(";")
+    parsed_params = dict(
+        param.strip().split("=", 1)
+        for param in params
+    )
+    parsed_params = {k: v.strip('"') for k, v in parsed_params.items()}
+    return main_value, parsed_params
+
+
+def process_record(record):
+    bucket_name = record["s3"]["bucket"]["name"]
+    file_key = record["s3"]["object"]["key"]
+    logger.info(f"Processing {file_key}")
+
+    response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
+    filename = response["Metadata"].get("mex-filename") or file_key
+    # TODO - this will read everything into memory - look at streaming instead
+    content = response["Body"].read().decode("utf-8")
 
+    content_type = response['ContentType']
+    media_type, content_type_params = parse_header_value(content_type)
 
-def ensure_dat_extension(file_name):
-    if '.' in file_name:
-        # Split the filename and extension
-        base_name, extension = file_name.rsplit('.', 1)
+    # Handle multipart content by parsing the filename and content from the first part
+    if media_type.startswith("multipart/"):
+        logger.info("Found multipart content")
+        boundary = content_type_params["boundary"]
+        parts = [
+            part.lstrip(f"--{boundary}")
+            for part in content.split(f"\r\n--{boundary}")
+            if part.strip() != "" and part.strip() != "--"
+        ]
+        if len(parts) > 1:
+            logger.warning(f"Got {len(parts)} parts, but will only process the first")
 
-        # Check if the extension is not 'dat'
-        if extension != 'dat':
-            file_name = f"{base_name}.dat"
-    else:
-        file_name += '.dat'
+        headers_str, content = parts[0].split("\r\n\r\n", 1)
+        headers = parse_headers(headers_str)
+        content_disposition = headers["Content-Disposition"]
+        _, content_disposition_params = parse_header_value(content_disposition)
+        filename = content_disposition_params.get("filename") or filename
 
-    return file_name
+    s3_client.put_object(Bucket=DESTINATION_BUCKET_NAME, Key=filename, Body=content.encode("utf-8"))
+
+
+def lambda_handler(event, _):
+    success = True
+
+    for record in event["Records"]:
+        try:
+            process_record(record)
+        except Exception:
+            logger.exception("Failed to process record")
+            success = False
+
+    return {
+        'statusCode': 200,
+        'body': 'Files converted and uploaded successfully!'
+    } if success else {
+        'statusCode': 500,
+        'body': 'Errors occurred during processing'
+    }