"""Utils.""" from src import lambda_exceptions def extract_triggered_key(cloudwatch_event): """Get S3 bucket and key from CloudWatch event. Args: cloudwatch_event (dict): AWS CloudWatch event data object. Returns: (str, str): Tuple of bucket and key names. """ try: if cloudwatch_event['detail-type'] == 'Object Created': detail = cloudwatch_event['detail'] bucket_name = detail['bucket']['name'] key = detail['object']['key'] return bucket_name, key elif cloudwatch_event['detail-type'] == ( 'GuardDuty Malware Protection Object Scan Result'): s3_object_details = ( cloudwatch_event['detail']['s3ObjectDetails']) bucket_name = s3_object_details['bucketName'] key = s3_object_details['objectKey'] return bucket_name, key else: # Default to old event structure for backward compatibility request_params = cloudwatch_event['detail']['requestParameters'] bucket_name = request_params['bucketName'] key = request_params['key'] return bucket_name, key except KeyError as e: raise lambda_exceptions.UnexpectedEventBody([str(e)])