DYNAMO DB USING PYTHON
3 min readOct 20, 2022
from ast import Notimport loggingimport timeimport jsonimport pickleimport datetimeimport timeimport stringimport randomimport sysimport threadingimport getoptimport boto3from boto3.dynamodb.conditions import Key, Attrimport oslogging.basicConfig(level=logging.INFO)orderbook_value = <json data># sts_client = boto3.client('sts')# KEY_ID = sts_session['Credentials']['AccessKeyId']# ACCESS_KEY = sts_session['Credentials']['SecretAccessKey']# TOKEN = sts_session['Credentials']['SessionToken']# print(KEY_ID)# dynamodb_client = boto3.client('dynamodb',# region_name='us-east-2',# aws_access_key_id=os.environ['KEY_ID'],# aws_secret_access_key=os.environ['ACCESS_KEY'],# aws_session_token=os.environ['TOKEN'])# dynamodb_client = boto3.client('dynamodb')dynamodb_client = boto3.resource('dynamodb')db_client = boto3.client('dynamodb')TABLE_NAME = "<table_name>"ob_table = dynamodb_client.Table(TABLE_NAME)def GenerateRandomString(lenstring):res = ''.join(random.choices(string.ascii_uppercase +string.digits, k=lenstring))# print result#print("The generated random string : " + str(res))return resdef ShowTables():try:#print(os.environ)tables = []for table in dynamodb_client.tables.all():print(table.name)tables.append(table)except Exception as e:print("exception occured in showing tables : "+str(e))else:return tablesdef StoreJSONDynamo(num):try :# 1 record is approx 1 kb in dummy data above . So to get 1 GB data on redis -> 1 * 1000 * 1000 num_of_recordsnum_of_records = numcounter = 0order_random_key = ""while counter < num_of_records:random_ucc = GenerateRandomString(6)order_random_key = order_type[0] + ":"+random_uccglobal orderbook_valuets = time.time()orderbook_value["header"]["ucc"] = random_uccorderbook_value["data"]["client_id"] = random_uccorderbook_value["header"]["timestamp"] = str(ts)ob_table.put_item(Item={"ucc": random_ucc,"data": orderbook_value})# read_dict = redis.get(order_random_key)# yourdict = pickle.loads(str.encode(read_dict))# print(yourdict)response = ob_table.get_item(Key={"ucc": random_ucc})print(response["Item"])counter = counter + 1except Exception as e:print("exception occurred in dynamo db {}".format(e))return counterdef StoreJSONDynamoBatch(num):try :# 1 record is approx 1 kb in dummy data above . So to get 1 GB data on redis -> 1 * 1000 * 1000 num_of_recordsnum_of_records = numcounter = 0num_pipeline = 100while counter < num_of_records:pipeline_counter = 0with ob_table.batch_writer() as batch:while pipeline_counter < num_pipeline and counter < num_of_records:random_ucc = GenerateRandomString(6)global orderbook_valuets = time.time()batch.put_item(Item={"ucc": random_ucc,"data": orderbook_value})pipeline_counter = pipeline_counter + 1counter = counter + 1except Exception as e:print("exception occurred in redis {}".format(e))return counterdef ScanDynamoDB():counter = 0response = ob_table.scan()data = response['Items']while 'LastEvaluatedKey' in response:response = ob_table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])for each in response['Items']:print(each)counter = counter + 1data.extend(response['Items'])return counter# counter = 0# scan = ob_table.scan()# for each in scan['Items']:# print(each)# counter = counter + 1# return counterdef CleanDynamoDB():scan = ob_table.scan()with ob_table.batch_writer() as batch:for each in scan['Items']:batch.delete_item(Key={'ucc': each['ucc']})def GetCommandLineOptions():argumentList = sys.argv[1:]# Optionsoptions = "si:"# Long optionslong_options = ["help", "insert=","pinsert=", "publish", "subscribe", "scan", "flush"]try:# Parsing argumentarguments, values = getopt.getopt(argumentList, options, long_options)# checking each argumentfor currentArgument, currentValue in arguments:if currentArgument in ("--help"):print ("python <script_name> <--help for help>/<--insert num for insertions>/<--publish for publish>/<--scan for scan the db>/<--flush for clearing the db>/<--subscribe for subscribe for db> ")elif currentArgument in ("--insert"):start = time.time()total_inserted = StoreJSONDynamo(int(currentValue))end = time.time()time_diff = end - startprint(f"time taken in sec : {str(time_diff)}")response = db_client.describe_table(TableName=TABLE_NAME)print("total items inside db : " + str(response["Table"]["ItemCount"]))# print("total db size : "+response) # number of keys inside the redis dbprint("total inserted count : "+str(total_inserted))elif currentArgument in ("--pinsert"):start = time.time()total_inserted = StoreJSONDynamoBatch(int(currentValue))end = time.time()time_diff = end - startprint(f"time taken in sec : {str(time_diff)}")response = db_client.describe_table(TableName=TABLE_NAME)print("total items inside db : " + str(response["Table"]["ItemCount"]))# print("total db size : "+response) # number of keys inside the redis dbprint("total inserted count : "+str(total_inserted))elif currentArgument in ("--scan"):start = time.time()total_count = ScanDynamoDB()print("total count : "+str(total_count))end = time.time()time_diff = end - startprint(f"time taken in sec : {str(time_diff)}")response = db_client.describe_table(TableName=TABLE_NAME)print("total items inside db : " + str(response["Table"]["ItemCount"]))elif currentArgument in ("--flush"):start = time.time()CleanDynamoDB()end = time.time()time_diff = end - startprint(f"time taken in sec : {str(time_diff)}")except getopt.error as err:# output error, and return with an error codeprint (str(err))if __name__ == '__main__':print("program started\n")#start = time.time()#CleanRedisDB()#StoreJSONRedis()#total_count = ScanRedis()#print("total count of data : "+ str(total_count))#StartPubSubThread()GetCommandLineOptions()#ShowTables()#StoreJSONDynamo(2)# end = time.time()# print("total db size : "+str(redis.dbsize())) # number of keys inside the redis db# time_diff = end - start# print(f"time taken in sec : {str(time_diff)}")print("\nprogram ends")
