File size: 1,861 Bytes
617260a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | import boto3
from utility import *
# initialize dynamodb instance
db_client = boto3.client(
"dynamodb",
aws_access_key_id = aws_access_key_id,
aws_secret_access_key = aws_secret_access_key
)
'''
dynamodb data operations
'''
# get the list of articles from articles table in dynamodb
def get_table(table_name:str):
result = db_client.scan(TableName = table_name,AttributesToGet = data_structure[table_name])
return db_list_to_py_list(result["Items"])
# add a new article to table articles in dynamodb, return error if failed
def post_item(table_name:str,item:dict):
try:
res = db_client.put_item(
TableName = table_name,
Item = py_dict_to_db_map(item)
)
except Exception as e:
return {"Error":e}
return res
# update an article in table articles in dynamodb, return error if failed
def put_item(table_name:str,item:dict):
try:
res = db_client.put_item(
TableName = table_name,
Item = py_dict_to_db_map(item)
)
except Exception as e:
return {"Error":e}
return res
# delete an article in table articles in dynamodb, return error if not found.
def delete_item(table_name:str,key:dict):
try:
res = db_client.delete_item(
TableName = table_name,
Key = py_dict_to_db_map(key)
)
except Exception as e:
return {"Error":e}
return res
def get_item(table_name:str,key:dict):
try:
res = db_client.get_item(
TableName = table_name,
Key = py_dict_to_db_map(key)
)
except Exception as e:
return {"Error":e}
return res
'''
dynamodb structure management
'''
def get_structure(table_name:str):
result = db_client.describe_table(TableName = table_name)
return result["Table"]["AttributeDefinitions"] |