KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use check_dynamodb method in localstack
Best Python code snippet using localstack_python
database.py
Source:database.py 
...35 )36 client.get_waiter('bucket_exists').wait(Bucket='blood-bank-dev-s3')37 logger.info('database.py check_s3 blood-bank-dev-s3 created END')38 return True39def check_dynamodb(resource=None, client=None):40 logger.info('database.py check_dynamodb BEGIN')41 if client is None or resource is None:42 resource, client = get_boto3_dynamodb()43 table_list = client.list_tables()['TableNames']44 if 'users' in table_list:45 logger.info('database.py check_dynamodb users table already exists')46 else:47 table = resource.create_table(48 TableName='users',49 KeySchema=[50 {51 'AttributeName': 'username',52 'KeyType': 'HASH'53 },54 ],55 AttributeDefinitions=[56 {57 'AttributeName': 'username',58 'AttributeType': 'S'59 },60 ],61 ProvisionedThroughput={62 'ReadCapacityUnits': 5,63 'WriteCapacityUnits': 564 }65 )66 table.meta.client.get_waiter('table_exists').wait(TableName='users')67 if 'user-donor' in table_list:68 logger.info('database.py check_dynamodb donor table already exists')69 else:70 table = resource.create_table(71 TableName='user-donor',72 KeySchema=[73 {74 'AttributeName': 'email',75 'KeyType': 'HASH'76 },77 ],78 AttributeDefinitions=[79 {80 'AttributeName': 'email',81 'AttributeType': 'S'82 },83 ],84 ProvisionedThroughput={85 'ReadCapacityUnits': 5,86 'WriteCapacityUnits': 587 }88 )89 table.meta.client.get_waiter('table_exists').wait(TableName='user-donor')90 if 'user-receiver' in table_list:91 logger.info('database.py check_dynamodb receiver table already exists')92 else:93 table = resource.create_table(94 TableName='user-receiver',95 KeySchema=[96 {97 'AttributeName': 'email',98 'KeyType': 'HASH'99 },100 ],101 AttributeDefinitions=[102 {103 'AttributeName': 'email',104 'AttributeType': 'S'105 },106 ],107 ProvisionedThroughput={108 'ReadCapacityUnits': 5,109 'WriteCapacityUnits': 5110 }111 )112 table.meta.client.get_waiter('table_exists').wait(TableName='user-receiver')113 if 'user-center' in table_list:114 logger.info('database.py check_dynamodb center table already exists')115 else:116 table = resource.create_table(117 TableName='user-center',118 KeySchema=[119 {120 'AttributeName': 'email',121 'KeyType': 'HASH'122 },123 ],124 AttributeDefinitions=[125 {126 'AttributeName': 'email',127 'AttributeType': 'S'128 },129 ],130 ProvisionedThroughput={131 'ReadCapacityUnits': 5,132 'WriteCapacityUnits': 5133 }134 )135 table.meta.client.get_waiter('table_exists').wait(TableName='user-center')136 # Wait until the table exists.137 if 'blood-requests' in table_list:138 logger.info('database.py check_dynamodb blood requests table already exists')139 else:140 table = resource.create_table(141 TableName='blood-requests',142 KeySchema=[143 {144 'AttributeName': 'uid',145 'KeyType': 'HASH'146 },147 ],148 AttributeDefinitions=[149 {150 'AttributeName': 'uid',151 'AttributeType': 'S'152 },153 ],154 ProvisionedThroughput={155 'ReadCapacityUnits': 5,156 'WriteCapacityUnits': 5157 }158 )159 table.meta.client.get_waiter('table_exists').wait(TableName='blood-requests')160 return True161def check_rds(resource=None, client=None):162 logger.info('database.py check_rds BEGIN')163 if client is None or resource is None:164 resource, client = get_boto3_rds()165 response = client.create_db_instance(166 AllocatedStorage=5,167 DBInstanceClass='db.t2.micro',168 DBInstanceIdentifier='mymysqlinstance',169 Engine='MySQL',170 MasterUserPassword='MyPassword',171 MasterUsername='MyUser',172 )173 client.get_waiter('db_instance_available').wait(DBInstanceIdentifier='mymysqlinstance')174 return True175def check_login_entity(email, password, entity):176 resource, client = get_boto3_dynamodb()177 print(entity)178 table = resource.Table('user-' + entity)179 response = table.get_item(180 Key={181 'email': email182 }183 )184 print(response)185 logger.info('database.py check_login_entity %s', response)186 if 'Item' in response and response['Item']['password'] == password:187 return True188 return False189def sign_up_entity(email, password, entity):190 logger.info('database.py sign_up_entity BEGIN')191 resource, client = get_boto3_dynamodb()192 table = resource.Table('user-' + entity)193 response = client.get_item(194 TableName='user-' + entity,195 Key={196 'email': {197 'S': email}198 }199 )200 if 'Item' in response:201 return False202 response = table.put_item(203 Item={204 'email': email,205 'password': password206 }207 )208 logger.info('database.py sign_up_entity %s', response)209 logger.info('database.py sign_up_entity END')210 return True211def get_donor_details(email):212 logger.info('database.py get_donor_details BEGIN')213 resource, client = get_boto3_dynamodb()214 table = resource.Table('user-donor')215 response = table.get_item(216 Key={217 'email': email218 }219 )220 item1 = response.get('Item')221 item1.pop('password')222 print(item1)223 logger.info('database.py get_donor_details END')224 return item1225def update_donor_details(form_data_dict, email):226 logger.info('database.py update_donor_details BEGIN')227 resource, client = get_boto3_dynamodb()228 table = resource.Table('user-donor')229 response = table.get_item(230 Key={231 'email': email232 }233 )234 item1 = response.get('Item')235 item2 = form_data_dict236 item1.update(item2)237 response = table.put_item(Item=item1)238 logger.info('database.py update_donor_details %s', response)239 logger.info('database.py update_donor_details END')240 return True241def get_donor_list():242 logger.info('database.py get_donor_list BEGIN')243 resource, client = get_boto3_dynamodb()244 table = resource.Table('user-donor')245 res = table.scan()246 print(res.get('Items'))247 logger.info('database.py get_donor_list END')248 return res.get('Items')249def get_blood_request_list():250 logger.info('database.py get_donor_blood_request_list BEGIN')251 resource, client = get_boto3_dynamodb()252 table = resource.Table('blood-requests')253 res = table.scan()254 print(res.get('Items'))255 logger.info('database.py get_donor_blood_request_list END')256 return res.get('Items')257def get_receiver_details(email):258 logger.info('database.py get_receiver_details BEGIN')259 resource, client = get_boto3_dynamodb()260 table = resource.Table('user-receiver')261 response = table.get_item(262 Key={263 'email': email264 }265 )266 item1 = response.get('Item')267 item1.pop('password')268 print(item1)269 logger.info('database.py get_receiver_details END')270 return item1271def update_receiver_details(form_data_dict, email):272 logger.info('database.py update_receiver_details BEGIN')273 resource, client = get_boto3_dynamodb()274 table = resource.Table('user-receiver')275 response = table.get_item(276 Key={277 'email': email278 }279 )280 item1 = response.get('Item')281 item2 = form_data_dict282 item1.update(item2)283 response = table.put_item(Item=item1)284 logger.info('database.py update_receiver_details %s', response)285 logger.info('database.py update_receiver_details END')286 return True287def update_receiver_blood_request(email, uid):288 logger.info('database.py update_receiver_blood_request BEGIN')289 resource, client = get_boto3_dynamodb()290 table = resource.Table('user-receiver')291 response = table.get_item(292 Key={293 'email': email294 }295 )296 item = response.get('Item')297 if 'blood_request_list' in item:298 blood_req_list = list(item.get('blood_request_list'))299 else:300 blood_req_list = []301 blood_req_list.append(uid)302 item['blood_request_list'] = blood_req_list303 response = table.put_item(Item=item)304 logger.info('database.py update_receiver_blood_request %s', response)305 logger.info('database.py update_receiver_blood_request END')306 return True307def get_blood_request_list_for_receiver(email):308 logger.info('database.py get_blood_request_list_for_receiver BEGIN')309 resource, client = get_boto3_dynamodb()310 table = resource.Table('user-receiver')311 response = table.get_item(312 Key={313 'email': email314 }315 )316 item = response.get('Item')317 if 'blood_request_list' in item:318 blood_req_list = list(item.get('blood_request_list'))319 else:320 return []321 blood_req_list_list = []322 for i in blood_req_list:323 blood_req_list_list.append(get_blood_request_from_uid(i))324 logger.info('database.py get_blood_request_list_for_receiver %s', blood_req_list_list)325 logger.info('database.py get_blood_request_list_for_receiver END')326 return blood_req_list_list327def get_blood_request_from_uid(uid):328 logger.info('database.py get_blood_request_from_uid BEGIN')329 resource, client = get_boto3_dynamodb()330 table = resource.Table('blood-requests')331 response = table.get_item(332 Key={333 'uid': uid334 }335 )336 item = response.get('Item')337 logger.info('database.py get_blood_request_from_uid %s', response)338 logger.info('database.py get_blood_request_from_uid END')339 return item340def get_center_details(email):341 logger.info('database.py get_center_details BEGIN')342 resource, client = get_boto3_dynamodb()343 table = resource.Table('user-center')344 response = table.get_item(345 Key={346 'email': email347 }348 )349 item1 = response.get('Item')350 item1.pop('password')351 print(item1)352 logger.info('database.py get_center_details END')353 return item1354def update_center_details(form_data_dict, email):355 logger.info('database.py update_receiver_details BEGIN')356 resource, client = get_boto3_dynamodb()357 table = resource.Table('user-center')358 response = table.get_item(359 Key={360 'email': email361 }362 )363 item1 = response.get('Item')364 item2 = form_data_dict365 item1.update(item2)366 response = table.put_item(Item=item1)367 logger.info('database.py update_center_details %s', response)368 logger.info('database.py update_center_details END')369 return True370def blood_request_allocate_to_donor(email, req_id, center_id):371 logger.info('database.py blood_request_allocate_to_donor BEGIN')372 resource, client = get_boto3_dynamodb()373 table = resource.Table('user-donor')374 response = table.get_item(375 Key={376 'email': email377 }378 )379 print(email, req_id, center_id)380 user = response.get('Item')381 print(response)382 if 'blood_request_list' in user:383 blood_req_list = list(user.get('blood_request_list'))384 else:385 blood_req_list = []386 blood_req_list.append(req_id)387 user['blood_request_list'] = blood_req_list388 response = table.put_item(Item=user)389 table = resource.Table('blood-requests')390 response = table.get_item(391 Key={392 'uid': req_id393 }394 )395 blood_req = response.get('Item')396 blood_req['allocated'] = True397 blood_req['allocated_user_id'] = email398 blood_req['allocated_center_id'] = center_id399 response = table.put_item(Item=blood_req)400 logger.info('database.py blood_request_allocate_to_donor %s', response)401 logger.info('database.py blood_request_allocate_to_donor END')402 return True403def initiate():404 logger.info('database.py initiate BEGIN')405 check_s3()406 check_dynamodb()407 # check_rds()408 logger.info('database.py initiate END')409 return True410def encrypt(message):411 try:412 if type(message) is str:413 message = message.encode()414 key = b'lggCcPdSdMHqH6c7P8C88B9KpysMoo5_zCjUTxhqPyA='415 return Fernet(key).encrypt(message).decode()416 except:417 return None418def decrypt(token):419 try:420 if type(token) is str:...
test_dynamodb.py
Source:test_dynamodb.py
...11 'endpoint_url': 'http://localhost:8000',12}13def is_dynamodb_running():14 """Test if a DynamoDB service is running locally"""15 async def check_dynamodb():16 session = aioboto3.Session()17 async with session.resource('dynamodb', **resource_kwargs) as resource:18 client = resource.meta.client19 await client.describe_limits()20 try:21 asyncio.run(check_dynamodb())22 return True23 except OSError:24 return False25pytestmark = [26 pytest.mark.asyncio,27 pytest.mark.skipif(28 not is_dynamodb_running(), reason='local DynamoDB service required for integration tests'29 ),30]31class TestDynamoDbCache(BaseStorageTest):32 storage_class = DynamoDbCache33 picklable = True34 init_kwargs = {35 'create_if_not_exists': True,...
test_aws_resources.py
Source:test_aws_resources.py
...19 connection=dynamodb_connection)20 yield table_name21 assert table.delete()22def test_check_dynamo_db(dynamodb_connection, dynamodb_table_name): # noqa23 result = check_dynamodb(dynamodb_connection, [dynamodb_table_name])24 assert 'dynamodb' in result25def test_check_dynamo_db_failed(dynamodb_connection): # noqa26 result = check_dynamodb(dynamodb_connection, ['not-exist'])27 assert 'dynamodb' in result28 assert result['dynamodb'] == 'fail'...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
