KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_dynamodb_table method in localstack

Best Python code snippet using localstack_python

test_exchange_repository.py

Source:test_exchange_repository.py

1from datetime import datetime, timezone, timedelta2from unittest.mock import patch3import pytest4from hallebarde.infrastructure import exchange_repository5@pytest.mark.usefixtures("setup_dynamodb_container")6@pytest.mark.usefixtures("get_dynamodb_table")7class TestExchangeRepositoryBasics:8    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')9    def test_save_should_create_readable_item(self, mock_get_table, an_exchange, get_dynamodb_table):10        # Given11        mock_get_table.return_value = get_dynamodb_table12        # When13        exchange_repository.save(an_exchange)14        # Then15        actual_exchange = exchange_repository.get(an_exchange.identifier)16        assert actual_exchange == an_exchange17    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')18    def test_get_should_return_none_if_no_id_is_matched(self, mock_get_table, an_exchange, get_dynamodb_table):19        # Given20        mock_get_table.return_value = get_dynamodb_table21        # When22        exchange_repository.save(an_exchange)23        # Then24        actual_exchange = exchange_repository.get('non-existent-id')25        assert actual_exchange is None26    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')27    def test_get_account_exchanges_should_return_only_account_exchanges(self, mock_get_table,28                                                                        two_exchanges_with_same_sub,29                                                                        an_exchange_with_different_sub,30                                                                        get_dynamodb_table):31        # Given32        mock_get_table.return_value = get_dynamodb_table33        for exchange in two_exchanges_with_same_sub:34            exchange_repository.save(exchange)35        exchange_repository.save(an_exchange_with_different_sub)36        # When37        actual_exchanges = exchange_repository.get_account_exchanges(two_exchanges_with_same_sub[0].sub)38        # Then39        assert actual_exchanges[0] in two_exchanges_with_same_sub40        assert actual_exchanges[1] in two_exchanges_with_same_sub41        assert len(actual_exchanges) == 242    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')43    def test_delete_should_delete_indicated_exchange(self, mock_get_table, an_exchange, get_dynamodb_table):44        # Given45        mock_get_table.return_value = get_dynamodb_table46        exchange_repository.save(an_exchange)47        # When48        exchange_repository.delete(an_exchange.identifier)49        # Then50        assert exchange_repository.get(an_exchange.identifier) is None51@pytest.mark.usefixtures("setup_dynamodb_container")52@pytest.mark.usefixtures("get_dynamodb_table")53class TestExchangeRepositoryGettingByToken:54    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')55    def test_get_identifier_from_token_should_return_identifier_associated_with_given_token(self, mock_get_table,56                                                                                            an_exchange,57                                                                                            get_dynamodb_table):58        # Given59        mock_get_table.return_value = get_dynamodb_table60        exchange_repository.save(an_exchange)61        # When62        identifier_from_upload_token = exchange_repository.get_identifier_from_token(63            upload_token=an_exchange.upload_token)64        identifier_from_download_token = exchange_repository.get_identifier_from_token(65            download_token=an_exchange.download_token)66        # Then67        assert identifier_from_upload_token == an_exchange.identifier68        assert identifier_from_download_token == an_exchange.identifier69    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')70    def test_get_identifier_from_token_should_return_none_if_exchange_does_not_exist(self, mock_get_table,71                                                                                     an_exchange,72                                                                                     get_dynamodb_table):73        # Given74        mock_get_table.return_value = get_dynamodb_table75        # When76        identifier_from_upload_token = exchange_repository.get_identifier_from_token(upload_token="fake_upload_token")77        identifier_from_download_token = exchange_repository.get_identifier_from_token(78            download_token="fake_download_token")79        # Then80        assert identifier_from_upload_token is None81        assert identifier_from_download_token is None82    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')83    def test_get_identifier_from_token_should_be_none_safe(self, mock_get_table, get_dynamodb_table):84        # Given85        mock_get_table.return_value = get_dynamodb_table86        # When87        identifier_from_upload_token = exchange_repository.get_identifier_from_token()88        # Then89        assert identifier_from_upload_token is None90    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')91    def test_revoke_upload_token(self, mock_get_table, an_exchange, get_dynamodb_table):92        # Given93        mock_get_table.return_value = get_dynamodb_table94        exchange_repository.save(an_exchange)95        # When96        exchange_repository.revoke_upload(an_exchange.identifier)97        actual_exchange = exchange_repository.get(an_exchange.identifier)98        # Then99        assert actual_exchange.revoked_upload is True100    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')101    def test_get_by_upload_token_should_return_exchange(self, mock_get_table, get_dynamodb_table, an_exchange):102        # Given103        mock_get_table.return_value = get_dynamodb_table104        exchange_repository.save(an_exchange)105        # When106        actual_exchange = exchange_repository.get_by_upload_token(an_exchange.upload_token)107        # Then108        assert actual_exchange == an_exchange109    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')110    def test_get_by_upload_token_should_return_none_if_no_exchange_exists(self, mock_get_table, get_dynamodb_table):111        # Given112        mock_get_table.return_value = get_dynamodb_table113        # When114        actual_exchange = exchange_repository.get_by_upload_token('anonexistingexchange')115        # Then116        assert actual_exchange is None117    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')118    def test_get_by_download_token_should_return_an_exchange(self, mock_get_table, get_dynamodb_table, an_exchange):119        # Given120        mock_get_table.return_value = get_dynamodb_table121        exchange_repository.save(an_exchange)122        # When123        actual_exchange = exchange_repository.get_by_download_token(an_exchange.download_token)124        # Then125        assert actual_exchange == an_exchange126    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')127    def test_get_by_download_token_should_return_none_if_no_exchange_exists(self, mock_get_table, get_dynamodb_table):128        # Given129        mock_get_table.return_value = get_dynamodb_table130        # When131        actual_exchange = exchange_repository.get_by_download_token('anonexistingexchange')132        # Then133        assert actual_exchange is None134@pytest.mark.usefixtures("setup_dynamodb_container")135@pytest.mark.usefixtures("get_dynamodb_table")136class TestExchangeRepositoryActionsBasedOnTime:137    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')138    def test_get_before_time_should_return_empty_list_if_no_exchanges_exist_before_this_time(self,139                                                                                             mock_get_table,140                                                                                             get_dynamodb_table,141                                                                                             generate_old_exchange):142        # Given143        mock_get_table.return_value = get_dynamodb_table144        week_before = datetime.now(timezone.utc) - timedelta(days=7)145        # When146        actual_exchanges = exchange_repository.get_before_time(week_before)147        # Then148        assert actual_exchanges == []149    @patch('hallebarde.infrastructure.exchange_repository._get_dynamodb_table')150    def test_get_before_time_should_return_only_exchanges_created_before_this_time(self, mock_get_table,151                                                                                   get_dynamodb_table, an_exchange,152                                                                                   generate_old_exchange):153        # Given154        mock_get_table.return_value = get_dynamodb_table155        time_before = datetime.now(timezone.utc) - timedelta(days=2)156        old_exchange = generate_old_exchange(days_before=3)157        exchange_repository.save(an_exchange)158        exchange_repository.save(old_exchange)159        # When160        actual_exchanges = exchange_repository.get_before_time(time_before)161        # Then...

bootstrap.py

Source:bootstrap.py

...15                            validators=[Required(), Length(min=1, max=10, message='10æåä»¥åã§å¥åãã¦ãã ãã'),16                                        Regexp('^[a-zA-Z0-9]+$', message='è±æ°åã§å¥åãã¦ãã ãã')17                                        ])18    submit = SubmitField('Submit')19def get_dynamodb_table(table_name):20    return boto3.resource('dynamodb').Table(table_name)21def check_user(user_name):22    table = get_dynamodb_table('zozo-image-user')23    response = table.get_item(24        Key={25            'user_id': user_name26        }27    )28    if 'Item' in response:29        return True30    else:31        return False32def add_user(user_name):33    table = get_dynamodb_table('zozo-image-user')34    table.put_item(35        Item={36            'user_id': user_name37        }38    )39def get_user_timeline(user_name):40    table = get_dynamodb_table('zozo-image-user-timeline')41    response = table.query(42        KeyConditionExpression=Key('user_id').eq(user_name),43        ScanIndexForward=False44    )45    return response['Items']46# resourceã ãè¿å´47def get_user_favorite():48    return get_dynamodb_table('zozo-image-user-favorite')49# resourceã ãè¿å´50def get_image_summary():51    return get_dynamodb_table('zozo-image-summary')52def set_display_content(timeline, favorite_res, summary_res):53    for item in timeline:54        # summaryãæ¤ç´¢55        summary_hit = summary_res.get_item(56            Key={57                'id': item['image_id'][0:8],58                'image_id': item['image_id']59            }60        )61        # favoriteãæ¤ç´¢62        favorite_hit = favorite_res.get_item(63            Key={64                'user_id': item['user_id'],65                'image_id': item['image_id']66            }67        )68        # summaryãåå¨ãã¦ããã°ããã®æå ±ãè¿½å 69        if 'Item' in summary_hit:70            item['category'] = summary_hit['Item']['category']71            item['brand'] = summary_hit['Item']['brand']72            item['sex'] = summary_hit['Item']['sex']73        # favoriteãåå¨ãã¦ããã°ããã©ã°ãç«ã¦ã74        if 'Item' in favorite_hit:75            item['flg'] = favorite_hit['Item']['flg'] # like(2) / dislike(1)76        else:77            item['flg'] = 078    print timeline79    return timeline80@app.route('/')81def index():82    user_name = session.get('user_name')83    if user_name is None or not check_user(user_name):84        return redirect(url_for('login'))85    else:86        timeline = get_user_timeline(user_name)87        favorite_resource = get_user_favorite()88        summary_resource = get_image_summary()89        items = set_display_content(timeline, favorite_resource, summary_resource)90        return render_template('index.html', user_name=user_name, items=items)91@app.route('/like', methods=['POST'])92def like():93    user_name = session.get('user_name')94    if user_name is None or not check_user(user_name):95        return redirect(url_for('login'))96    if request.method == 'POST' and request.headers['Content-Type'] == 'application/json':97        table = get_dynamodb_table('zozo-image-user-favorite')98        response = table.update_item(99            Key={100                'user_id': user_name,101                'image_id': request.json102            },103            AttributeUpdates={104                'flg': {105                    'Action': 'PUT',106                    'Value': 2107                },108                'updatetime': {109                    'Action': 'PUT',110                    'Value': datetime.now().strftime("%Y%m%d%H%M%S%f")111                }112            }113        )114        return jsonify(response)115    else:116        return redirect(url_for('index'))117@app.route('/unlike', methods=['POST'])118def unlike():119    user_name = session.get('user_name')120    if user_name is None or not check_user(user_name):121        return redirect(url_for('login'))122    if request.method == 'POST' and request.headers['Content-Type'] == 'application/json':123        table = get_dynamodb_table('zozo-image-user-favorite')124        response = table.update_item(125            Key={126                'user_id': user_name,127                'image_id': request.json128            },129            AttributeUpdates={130                'flg': {131                    'Action': 'PUT',132                    'Value': 1133                },134                'updatetime': {135                    'Action': 'PUT',136                    'Value': datetime.now().strftime("%Y%m%d%H%M%S%f")137                }...

dynamodb_support.py

Source:dynamodb_support.py

...6from chalice import ChaliceViewError, BadRequestError7def put_item(table_name: str, item, dynamodb=None):8    try:9        print("Inserting into table {}, value {}", table_name, json.dumps(item))10        get_dynamodb_table(table_name, dynamodb).put_item(Item=item)11    except ClientError as ce:12        traceback.print_exc()13        print(ce.response['Error']['Message'])14    except Exception as e:15        traceback.print_exc()16        raise e17def update_item(table_name: str, key, expression, attribute_names, attribute_values, dynamodb=None):18    try:19        return get_dynamodb_table(table_name, dynamodb).update_item(Key=key, UpdateExpression=expression,20                                                                    ExpressionAttributeNames=attribute_names,21                                                                    ExpressionAttributeValues=attribute_values,22                                                                    ReturnValues="UPDATED_NEW")23    except ClientError as ce:24        traceback.print_exc(ce)25        print(ce.response['Error']['Message'])26        raise ChaliceViewError(ce.response['Error']['Message'])27    except Exception as e:28        traceback.print_exc()29        raise e30def delete_item(table_name: str, key: dict, dynamodb=None):31    try:32        return get_dynamodb_table(table_name, dynamodb).delete_item(Key={'id': key})33    except ClientError as ce:34        traceback.print_exc(ce)35        print(ce.response['Error']['Message'])36        raise ChaliceViewError(ce.response['Error']['Message'])37    except Exception as e:38        traceback.print_exc()39        raise e40def query(table_name: str, condition_expression, dynamodb=None):41    try:42        return get_dynamodb_table(table_name, dynamodb).query(**condition_expression)['Items']43    except ClientError as ce:44        traceback.print_exc()45        print(ce.response['Error']['Message'])46        raise ChaliceViewError(ce.response['Error']['Message'])47    except Exception as e:48        traceback.print_exc()49        raise e50def get_by_id(table_name: str, key: str, dynamodb=None):51    try:52        result = get_dynamodb_table(table_name, dynamodb).get_item(Key={53            'id': key54        })55        if 'Item' in result:56            return result['Item']57        raise BadRequestError('Id {} not found'.format(key))58    except ClientError as ce:59        traceback.print_exc()60        print(ce.response['Error']['Message'])61        raise ChaliceViewError(ce.response['Error']['Message'])62    except Exception as e:63        traceback.print_exc()64        raise e65def get_all_items(table_name: str, dynamodb=None):66    pass67def get_dynamodb_table(table_name: str, dynamodb):68    if not dynamodb:69        dynamodb = boto3.resource('dynamodb', region_name='us-west-1', endpoint_url=os.getenv('dynamodb_url'))...

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.