Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.