File size: 3,348 Bytes
b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e 0d5eb6c b426b8e | 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | 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 items from a table in dynamodb
def get_table(table_name:str):
'''
get the list of items from table in dynamodb
Args:
table_name (str): the name of the table in dynamodb
Returns:
list: a list of items in the table
'''
result = db_client.scan(TableName = table_name,AttributesToGet = data_structure[table_name]["fields"])
return [db_map_to_py_dict(r) for r in result["Items"]]
# add a new item to a table in dynamodb, return error if failed
def post_item(table_name:str,item:dict):
'''
add a new item to table in dynamodb, return error if failed
Args:
table_name (str): the name of the table in dynamodb
item (dict): the item to be added to the table
Returns:
dict: the result of the operation'''
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 item in a table in dynamodb, return error if failed
def put_item(table_name:str,item:dict):
'''
update an item in table in dynamodb, return error if failed
Args:
table_name (str): the name of the table in dynamodb
item (dict): the item to be updated to the table
Returns:
dict: the result of the operation
'''
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 item in a table in dynamodb, return error if not found.
def delete_item(table_name:str,key:dict):
'''
delete an item in table in dynamodb, return error if not found.
Args:
table_name (str): the name of the table in dynamodb
key (dict): the key of the item to be deleted
Returns:
dict: the result of the operation
'''
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):
'''
get an item in table in dynamodb, return error if not found.
Args:
table_name (str): the name of the table in dynamodb
key (dict): the key of the item to be deleted
Returns:
dict: the result of the operation
'''
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):
'''
get the structure of a table in dynamodb
Args:
table_name (str): the name of the table in dynamodb
Returns:
dict: the structure of the table
'''
result = db_client.describe_table(TableName = table_name)
return result["Table"]["AttributeDefinitions"] |