Last active
March 17, 2019 16:27
-
-
Save snimmagadda1/b33613f918e468c8193fb29a72264856 to your computer and use it in GitHub Desktop.
Create an AWS DynamoDB item from a python dictionary, initialize a table, put an item into DynamoDB (using boto3)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
import boto3 | |
import json | |
import decimal | |
from botocore.exceptions import ClientError | |
logger = logging.getLogger() | |
# Format outputs | |
class DecimalEncoder(json.JSONEncoder): | |
def default(self, o): | |
if isinstance(o, decimal.Decimal): | |
if o % 1 > 0: | |
return float(o) | |
else: | |
return int(o) | |
return super(DecimalEncoder, self).default(o) | |
def initialize_dynamoDB(tableName): | |
"""Initialize logging and the DynamoDB table. Assumes AWS CLI is configured. | |
:param tableName: Table to load (string). | |
:return dynamoTable: AWS DynamoDB table. | |
""" | |
logger.setLevel(logging.INFO) | |
dynamoTable = boto3.resource('dynamodb').Table(tableName) | |
logger.info("Loaded table: " + tableName) | |
return dynamoTable | |
def dict_2_dynamoDB_item(raw): | |
"""Make a DynamoDB item from a python dictionary. | |
:param raw: The raw data to create the item (dict). | |
:return result: Formatted DynamoDB item (dict). | |
""" | |
if type(raw) is dict: | |
result = {} | |
for k, v in raw.items(): | |
if type(v) is str: | |
if not v: | |
result[k] = "NULL" | |
else: | |
result[k] = v | |
if type(v) is int: | |
result[k] = str(v) | |
if type(v) is dict: | |
result[k] = dict_2_dynamoDB_item(v) | |
if type(v) is list: | |
result[k] = [] | |
for i in v: | |
result[k].append(dict_2_dynamoDB_item(i)) | |
return result | |
def upload_item(item, dynamoTable ): | |
"""Put an item into an AWS DynamoDB table. | |
:param item: The item to store (dict) | |
:param dynamoTable: The table to store in. | |
:return: void. | |
""" | |
try: | |
response = dynamoTable.put_item( | |
Item=item | |
) | |
except ClientError as e: | |
print(e.response['Error']['Message']) | |
else: | |
print("PutItem succeeded:") | |
print(json.dumps(response, indent=4, cls=DecimalEncoder)) | |
return |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment