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 Github

copy

Full Screen

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...

Full Screen

Full Screen

bootstrap.py

Source: bootstrap.py Github

copy

Full Screen

...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 }...

Full Screen

Full Screen

dynamodb_support.py

Source: dynamodb_support.py Github

copy

Full Screen

...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'))...

Full Screen

Full Screen

Blogs

Check out the latest blogs from LambdaTest on this topic:

Agile in Distributed Development – A Formula for Success

Agile has unquestionable benefits. The mainstream method has assisted numerous businesses in increasing organizational flexibility as a result, developing better, more intuitive software. Distributed development is also an important strategy for software companies. It gives access to global talent, the use of offshore outsourcing to reduce operating costs, and round-the-clock development.

How To Find Hidden Elements In Selenium WebDriver With Java

Have you ever struggled with handling hidden elements while automating a web or mobile application? I was recently automating an eCommerce application. I struggled with handling hidden elements on the web page.

Test Managers in Agile – Creating the Right Culture for Your SQA Team

I was once asked at a testing summit, “How do you manage a QA team using scrum?” After some consideration, I realized it would make a good article, so here I am. Understand that the idea behind developing software in a scrum environment is for development teams to self-organize.

Running Tests In Cypress With GitHub Actions [Complete Guide]

In today’s tech world, where speed is the key to modern software development, we should aim to get quick feedback on the impact of any change, and that is where CI/CD comes in place.

Oct’22 Updates: New Analytics And App Automation Dashboard, Test On Google Pixel 7 Series, And More

Hey everyone! We hope you had a great Hacktober. At LambdaTest, we thrive to bring you the best with each update. Our engineering and tech teams work at lightning speed to deliver you a seamless testing experience.

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful