|
| 1 | +import json |
| 2 | +import boto3 |
| 3 | +import urllib.parse |
| 4 | +from typing import Dict, Any, Optional |
| 5 | +import logging |
| 6 | + |
| 7 | +# Configure logging |
| 8 | +logger = logging.getLogger() |
| 9 | +logger.setLevel(logging.INFO) |
| 10 | + |
| 11 | +# Initialize AWS clients |
| 12 | +dynamodb = boto3.client("dynamodb") |
| 13 | +s3 = boto3.client("s3") |
| 14 | + |
| 15 | + |
| 16 | +def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: |
| 17 | + """ |
| 18 | + Lambda function to handle S3 upload events and update DynamoDB. |
| 19 | +
|
| 20 | + Expects S3 object metadata: |
| 21 | + - dynamoTable: DynamoDB table name |
| 22 | + - dynamoPrimaryKey: JSON string of primary key |
| 23 | + - dynamoAttribute: Target attribute name to set with value from pending attribute |
| 24 | + - dynamoPendingAttributeName: Source pending attribute name to remove |
| 25 | + """ |
| 26 | + try: |
| 27 | + # Process each S3 event record |
| 28 | + for record in event["Records"]: |
| 29 | + process_s3_record(record) |
| 30 | + |
| 31 | + return { |
| 32 | + "statusCode": 200, |
| 33 | + "body": json.dumps("Successfully processed S3 events"), |
| 34 | + } |
| 35 | + |
| 36 | + except Exception as e: |
| 37 | + logger.error(f"Error processing S3 event: {str(e)}", exc_info=True) |
| 38 | + raise |
| 39 | + |
| 40 | + |
| 41 | +def process_s3_record(record: Dict[str, Any]) -> None: |
| 42 | + """Process a single S3 event record.""" |
| 43 | + |
| 44 | + # Extract S3 event details |
| 45 | + bucket = record["s3"]["bucket"]["name"] |
| 46 | + key = urllib.parse.unquote_plus(record["s3"]["object"]["key"]) |
| 47 | + |
| 48 | + logger.info(f"Processing upload for bucket={bucket}, key={key}") |
| 49 | + |
| 50 | + # Get object metadata |
| 51 | + metadata = get_object_metadata(bucket, key) |
| 52 | + |
| 53 | + if not metadata: |
| 54 | + logger.warning(f"No metadata found for object {key}. Skipping DynamoDB update.") |
| 55 | + return |
| 56 | + |
| 57 | + # Extract required metadata fields |
| 58 | + dynamo_table = metadata.get("dynamotable") |
| 59 | + dynamo_primary_key_json = metadata.get("dynamoprimarykey") |
| 60 | + dynamo_attribute = metadata.get("dynamoattribute") |
| 61 | + dynamo_pending_attribute = metadata.get("dynamopendingattributename") |
| 62 | + |
| 63 | + # Validate required metadata - exit early if any are missing |
| 64 | + if not dynamo_table: |
| 65 | + logger.warning(f"Missing dynamoTable metadata for {key}") |
| 66 | + return |
| 67 | + |
| 68 | + if not dynamo_primary_key_json: |
| 69 | + logger.warning(f"Missing dynamoPrimaryKey metadata for {key}") |
| 70 | + return |
| 71 | + |
| 72 | + if not dynamo_attribute: |
| 73 | + logger.warning(f"Missing dynamoAttribute metadata for {key}") |
| 74 | + return |
| 75 | + |
| 76 | + if not dynamo_pending_attribute: |
| 77 | + logger.warning(f"Missing dynamoPendingAttributeName metadata for {key}") |
| 78 | + return |
| 79 | + |
| 80 | + # Parse primary key |
| 81 | + try: |
| 82 | + primary_key = json.loads(dynamo_primary_key_json) |
| 83 | + except json.JSONDecodeError as e: |
| 84 | + logger.error(f"Failed to parse dynamoPrimaryKey JSON: {e}") |
| 85 | + return |
| 86 | + |
| 87 | + # Update DynamoDB - all variables are guaranteed to be strings now |
| 88 | + update_dynamodb( |
| 89 | + table_name=dynamo_table, |
| 90 | + primary_key=primary_key, |
| 91 | + target_attribute=dynamo_attribute, |
| 92 | + pending_attribute=dynamo_pending_attribute, |
| 93 | + ) |
| 94 | + |
| 95 | + logger.info(f"Successfully updated DynamoDB for {key}") |
| 96 | + |
| 97 | + |
| 98 | +def get_object_metadata(bucket: str, key: str) -> Optional[Dict[str, str]]: |
| 99 | + """Retrieve metadata from S3 object.""" |
| 100 | + try: |
| 101 | + response = s3.head_object(Bucket=bucket, Key=key) |
| 102 | + return response.get("Metadata", {}) |
| 103 | + except Exception as e: |
| 104 | + logger.error(f"Error getting metadata for {bucket}/{key}: {str(e)}") |
| 105 | + return None |
| 106 | + |
| 107 | + |
| 108 | +def update_dynamodb( |
| 109 | + table_name: str, |
| 110 | + primary_key: Dict[str, str], |
| 111 | + target_attribute: str, |
| 112 | + pending_attribute: str, |
| 113 | +) -> None: |
| 114 | + """ |
| 115 | + Update DynamoDB item, moving value from pending attribute to target attribute. |
| 116 | +
|
| 117 | + Args: |
| 118 | + table_name: DynamoDB table name |
| 119 | + primary_key: Primary key as dict (e.g., {"requestId": "123", "createdAt#status": "..."}) |
| 120 | + target_attribute: The confirmed attribute name (e.g., "attachmentS3key") |
| 121 | + pending_attribute: The pending attribute name (e.g., "pendingAttachmentS3key") |
| 122 | + """ |
| 123 | + |
| 124 | + # Convert primary key to DynamoDB format |
| 125 | + dynamo_key = {k: {"S": v} for k, v in primary_key.items()} |
| 126 | + |
| 127 | + try: |
| 128 | + # Build update expression to move pending attribute value to target attribute |
| 129 | + # SET target = pending, REMOVE pending |
| 130 | + update_expression = "SET #target = #pending REMOVE #pending" |
| 131 | + |
| 132 | + expression_attribute_names = { |
| 133 | + "#target": target_attribute, |
| 134 | + "#pending": pending_attribute, |
| 135 | + } |
| 136 | + |
| 137 | + # Condition: pending attribute should exist and equal the uploaded s3 key |
| 138 | + condition_expression = ( |
| 139 | + "attribute_exists(#pending) AND #pending = :expected_s3key" |
| 140 | + ) |
| 141 | + |
| 142 | + dynamodb.update_item( |
| 143 | + TableName=table_name, |
| 144 | + Key=dynamo_key, |
| 145 | + UpdateExpression=update_expression, |
| 146 | + ExpressionAttributeNames=expression_attribute_names, |
| 147 | + ConditionExpression=condition_expression, |
| 148 | + ReturnValues="ALL_NEW", |
| 149 | + ) |
| 150 | + |
| 151 | + logger.info( |
| 152 | + f"Updated DynamoDB table={table_name}, " |
| 153 | + f"key={primary_key}, " |
| 154 | + f"moved value from {pending_attribute} to {target_attribute}" |
| 155 | + ) |
| 156 | + |
| 157 | + except dynamodb.exceptions.ConditionalCheckFailedException: |
| 158 | + logger.info( |
| 159 | + f"Skipping update for {table_name} with key {primary_key}. " |
| 160 | + f"This is expected if the file was already confirmed or uploaded without metadata." |
| 161 | + ) |
| 162 | + except Exception as e: |
| 163 | + logger.error( |
| 164 | + f"Error updating DynamoDB table={table_name}, key={primary_key}: {str(e)}", |
| 165 | + exc_info=True, |
| 166 | + ) |
| 167 | + raise |
0 commit comments