Skip to content

Instantly share code, notes, and snippets.

@snimmagadda1
Last active March 17, 2019 16:27
Show Gist options
  • Save snimmagadda1/b33613f918e468c8193fb29a72264856 to your computer and use it in GitHub Desktop.
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)
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