Source code for lambdas.storepick_stream_to_s3
import traceback
import boto3
import botocore
import logging
import json
import datetime
from pymorri.util.application_dependencies import ApplicationDependencies
from pymorri.util.serverless_utils import resolve_env
logging.basicConfig(level=logging.INFO, format='%(levelname)s :%(name)s: %(message)s')
logger = logging.getLogger('lambdas.storepick_stream_to_s3')
configuration = None
[docs]def get_config(name: str) -> dict:
"""
Get the config from the config service. Uses global variable for caching and reducing network calls.
Params:
name (string): Name of the config
Returns:
configuration (dict): Config dictionary
"""
global configuration
if configuration:
return configuration
dependencies = ApplicationDependencies(config_file=name)
configuration = resolve_env(dependencies.get_config())
return configuration
[docs]def get_current_datetime() -> dict:
"""This function generated the current date & time applying appropriate formatting for the s3 path
Returns:
(dict) : Dictionary containing current date & time
Example: {'date': '20210510', 'time': '11:56:02'}
"""
today = datetime.datetime.now()
date = today.date().strftime("%Y%m%d")
timestamp = today.strftime("%H:%M:%S")
return {"date": date, "time": timestamp}
[docs]def create_s3_path(root_folder, seq_num):
"""This function makes creates the path in which the file will be saved within the s3 bucket
Params:
root_folder (string) : root folder of s3 bucket
seq_num (string): SequenceNumber from the event received from dynamodb
Returns:
s3_path (string) : the full file path of where the data will be written to within s3
"""
curr_datetime = get_current_datetime()
file_name = seq_num + "-" + curr_datetime['time'] + ".json"
s3_path = root_folder + "/" + curr_datetime['date'] + "/" + file_name
return s3_path
[docs]def set_component_logging(config):
"""
This method is used to set the logging level for BigQuery Writer and Pymorri.
"""
logger.setLevel(level=logging.getLevelName(config['OSP_PICKS_TO_KAFKA_LOG_LVL']))
logging.getLogger('pymorri').setLevel(
level=logging.getLevelName(config['OSP_PICKS_TO_KAFKA_PYMORRI_LOG_LVL']))
[docs]def lambda_handler(event, context):
"""
This Lambda receives events from DynamoDB streams. The list of data from within the event will be saved to s3.
Params:
event: Event from DynamoDB streams
context: Context from DynamoDB streams
Returns:
response (json): The response meta data
"""
config = get_config('osp-picks-to-kafka.json')
set_component_logging(config)
logger.debug("config %s", config)
logger.debug("context %s", context)
S3_BUCKET_NAME = config['S3_BUCKET_NAME']
S3_ROOT_FOLDER = config['S3_ROOT_FOLDER']
try:
records = event['Records']
seq_num = records[0]['dynamodb']['SequenceNumber']
logger.info(f"Processing started from record {seq_num}")
s3_path = create_s3_path(S3_ROOT_FOLDER, seq_num)
logger.info(f"Attempting to write file {s3_path}")
s3 = boto3.client("s3")
s3.put_object(Key=s3_path, Body=str(json.dumps(records)), Bucket=S3_BUCKET_NAME)
except botocore.exceptions.ClientError as e:
traceback.print_exc()
logger.error(f"OSP_DYNAMODB_TO_S3_ERROR - ClientError - Unexpected client error. Please check config and "
f"bucket exists. {e}")
raise e
except Exception as e:
traceback.print_exc()
logger.critical(f"OSP_DYNAMODB_TO_S3_ERROR_CRITICAL An Unexpected error occurred. Please check configuration. {e}")
raise e
if __name__ == '__main__':
dynamodb_stream_event = {
"Records": [
{
"eventID": "f373a5d3c645777f89141890f3de5f6b",
"eventName": "MODIFY",
"eventVersion": "1.1",
"eventSource": "aws:dynamodb",
"awsRegion": "eu-west-1",
"dynamodb": {
"ApproximateCreationDateTime": 1622903039,
"Keys": {
"pickTaskId": {
"S": "3e3a63fb-bc43-4cb5-8695-3722e3253ac9"
}
},
"NewImage": {
"eventId": {
"S": "e56bbe76-3b5c-4388-a329-82178530ee52"
},
"quantityPicked": {
"N": "3"
},
"ttlattribute": {
"N": "1654525439"
},
"productId": {
"S": "102793671"
},
"substitutions": {
"L": []
},
"orderId": {
"S": "98519"
},
"pickLocationId": {
"S": "U-RX100-3"
},
"pickTaskId": {
"S": "3e3a63fb-bc43-4cb5-8695-3722e3253ac9"
},
"siteId": {
"S": "132"
},
"quantityOrdered": {
"N": "7"
},
"lastUpdatedDateTime": {
"S": "2021-06-05 14:23:59.445184"
},
"timestamp": {
"S": "2021-05-26T09:06:29.000Z"
}
},
"OldImage": {
"eventId": {
"S": "e56bbe76-3b5c-4388-a329-82178530ee52"
},
"quantityPicked": {
"N": "3"
},
"ttlattribute": {
"N": "1654445554"
},
"productId": {
"S": "102793671"
},
"substitutions": {
"L": []
},
"orderId": {
"S": "21077"
},
"pickLocationId": {
"S": "U-RX100-3"
},
"pickTaskId": {
"S": "3e3a63fb-bc43-4cb5-8695-3722e3253ac9"
},
"siteId": {
"S": "132"
},
"quantityOrdered": {
"N": "7"
},
"lastUpdatedDateTime": {
"S": "2021-06-04 16:12:34.487837"
},
"timestamp": {
"S": "2021-05-26T09:06:29.000Z"
}
},
"SequenceNumber": "26406200000000021183724368",
"SizeBytes": 638,
"StreamViewType": "NEW_AND_OLD_IMAGES"
},
"eventSourceARN": "arn:aws:dynamodb:eu-west-1:928369729144:table/dev.xxwmm.storepick.osp-eh-v2/stream/2021-05-31T10:27:10.516"
}
]
}
lambda_handler(dynamodb_stream_event, "Test")