AWS Lambda with python example — Inserting data into dynamodb table from s3

AWS lambda is a serverless computing service . Please refer below link for more information about AWS lambda and for creating your first lambda function in python.

AWS Lambda with python examples

Lets start discussing about an another example — Inserting data items into a dynamodb table from a csv file, which is stored in an s3 bucket.

Step 1 : Create the dynamodb table

Here I have created a table employee with two attributes username and lastname.

employee table in — Dynamodb

Step 2 : Create an IAM role for the lambda function.

This particular example requires three different AWS services S3, Dynamodb and CloundWatch. So the IAM role should have the following polices attached to it.

IAM role for the lambda function

Step 3 : Create the Lambda function. (Refer the first link for the configuration)

Lambda function design

Ok Let’s discuss about the lambda function example in detail.

This lambda function would get invoked when a csv file upload event happens in the configured S3 bucket. Once this function gets triggered, the lambda_handler() function gets the event and context objects. This particular lambda lambda_handler() function fetches the csv file and converts all the records into the form of a dictionary and adds it into a list. The column names are used as keys in the record dictionary.

# lambda_handler is the main function in lambda function
def lambda_handler(event,context):
    source_bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.unquote_plus(event['Records'][0]['s3']['object']['key'])
    copy_source = {'Bucket':source_bucket , 'Key':key}
    print(event)
    
    #just print function
    print("Log stream name : ", context.log_stream_name)
    print("Log group name : ", context.log_group_name)
    print("Request Id:", context.aws_request_id)
    print("Mem. limit(MB): ", context.memory_limit_in_mb)
    
    try:
        print("Using waiter to waiting for object to persist thru s3 service")
        waiter = s3.get_waiter('object_exists')
        waiter.wait(Bucket=source_bucket, Key=key)
        print("Accessing the receied file and reading the same")
        bucket = tests3.Bucket(u'awslambdas3test2')
        obj = bucket.Object(key='inputfile.csv')
        response = obj.get()
        print("response from file object")
        print(response)
        lines = response['Body'].read().split()
        print(response['Body'].read())
        
        # creating a list of dictionaries from the csv file records
        recList = list()
        i = 0
        while i < len(lines):
            record  = {}
            record['username'] = lines[i]
            record['lastname'] = lines[i+1]
            print(record)
            recList.append(record)
            i = i+2
        print(recList)    
        insert_data(recList)
        
    except Exception as e:
        print(e)
        print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, source_bucket))
        raise e

Once all the records in the csv file are converted into list, it will pass to the insert_data function. insert_data() function takes control of employee table first, then iterates the record list and inserts it into the table using put_item function.

# insert_data function for inserting data into dynamodb table
def insert_data(recList):
    table = dynamodb.Table('employee')
    for i in range(len(recList)):
        record = recList[i]
        table.put_item(
            Item={
                'username': record['username'],
                'lastname': record['lastname']
            }
        )

Now you have completed the lambda function for Inserting data items into a dynamodb table from a csv file, which is stored in an s3 bucket. Save the function and upload the csv file into the configured s3 bucket. Now you can find the csv file contents in the dynamo db table.

There might be different approaches available for this problem, but I have done this way and its working fine for me. Please find the git location for the code https://github.com/stephinmon/stephin_dev/blob/master/dynamodbinsertfroms3.py .

Thanks for reading and hope this article is helpful for you !!

Please comment your valuable suggestions in the comment box . Please reach me for any queries via my email stephinmon.antony@gmail.com .