File size: 2,145 Bytes
43de500 c6e60b1 43de500 3337920 43de500 004db8a 43de500 3337920 43de500 3337920 43de500 45f7b67 3337920 45f7b67 43de500 3337920 43de500 |
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 72 73 74 75 76 77 78 79 80 81 82 |
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,
region_name = "us-east-1"
)
'''
dynamodb data operations
'''
# get the list of articles from articles table in dynamodb
@terminal_print
def get_table(table_name:str):
result = db_client.scan(TableName = table_name)
items = result["Items"]
while "LastEvaluatedKey" in result:
result = db_client.scan(TableName = table_name,ExclusiveStartKey = result["LastEvaluatedKey"])
items.extend(result["Items"])
return [db_map_to_py_dict(r) for r in 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
@terminal_print
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.
@terminal_print
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
'''
'''
@terminal_print
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
'''
@terminal_print
def get_structure(table_name:str):
result = db_client.describe_table(TableName = table_name)
return result["Table"]["AttributeDefinitions"] |