Best Python code snippet using localstack_python
start_state_machine.py
Source: start_state_machine.py
1#!/usr/bin/env python2# Copyright 2016-2020 Workiva Inc.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15#16# start_state_machine.py17#18# Script that injects an initial event into Kinesis.19# system imports20import argparse21import json22import logging23import sys24# library imports25# application imports26from aws_lambda_fsm.aws import get_connection27from aws_lambda_fsm.aws import get_arn_from_arn_string28from aws_lambda_fsm.aws import validate_config29from aws_lambda_fsm.client import start_state_machine30from aws_lambda_fsm.client import start_state_machines31from aws_lambda_fsm.constants import STATE32from aws_lambda_fsm.constants import SYSTEM_CONTEXT33from aws_lambda_fsm.constants import AWS_KINESIS34from aws_lambda_fsm.constants import AWS35from aws_lambda_fsm.serialization import json_loads_additional_kwargs36import settings37# setup the command line args38parser = argparse.ArgumentParser(description='Starts a state machine.')39parser.add_argument('--machine_name')40parser.add_argument('--checkpoint_shard_id')41parser.add_argument('--checkpoint_sequence_number')42parser.add_argument('--initial_context')43parser.add_argument('--num_machines', type=int, default=1)44parser.add_argument('--log_level', default='INFO')45parser.add_argument('--boto_log_level', default='INFO')46parser.add_argument('--correlation_id')47args = parser.parse_args()48logging.basicConfig(49 format='[%(levelname)s] %(asctime)-15s %(message)s',50 level=int(args.log_level) if args.log_level.isdigit() else args.log_level,51 datefmt='%Y-%m-%d %H:%M:%S'52)53logging.getLogger('boto3').setLevel(args.boto_log_level)54logging.getLogger('botocore').setLevel(args.boto_log_level)55validate_config()56if args.num_machines > 1:57 # start things off58 context = json.loads(args.initial_context or "{}", **json_loads_additional_kwargs())59 current_state = current_event = STATE.PSEUDO_INIT60 start_state_machines(args.machine_name,61 [context] * args.num_machines,62 current_state=current_state,63 current_event=current_event)64 exit(0)65# checkpoint specified, so start with a context saved to the kinesis stream66if args.checkpoint_shard_id and args.checkpoint_sequence_number:67 # setup connections to AWS68 kinesis_stream_arn = getattr(settings, args.kinesis_stream_arn)69 logging.info('Kinesis stream ARN: %s', kinesis_stream_arn)70 logging.info('Kinesis endpoint: %s', settings.ENDPOINTS.get(AWS.KINESIS))71 if get_arn_from_arn_string(kinesis_stream_arn).service != AWS.KINESIS:72 logging.fatal("%s is not a Kinesis ARN", kinesis_stream_arn)73 sys.exit(1)74 kinesis_conn = get_connection(kinesis_stream_arn)75 kinesis_stream = get_arn_from_arn_string(kinesis_stream_arn).slash_resource()76 logging.info('Kinesis stream: %s', kinesis_stream)77 # create a shard iterator for the specified shard and sequence number78 shard_iterator = kinesis_conn.get_shard_iterator(79 StreamName=kinesis_stream,80 ShardId=args.checkpoint_shard_id,81 ShardIteratorType=AWS_KINESIS.AT_SEQUENCE_NUMBER,82 StartingSequenceNumber=args.checkpoint_sequence_number83 )[AWS_KINESIS.ShardIterator]84 # get the record that has the last successful state85 records = kinesis_conn.get_records(86 ShardIterator=shard_iterator,87 Limit=188 )89 if records:90 context = json.loads(records[AWS_KINESIS.Records][0][AWS_KINESIS.DATA], **json_loads_additional_kwargs())91 current_state = context.get(SYSTEM_CONTEXT.CURRENT_STATE)92 current_event = context.get(SYSTEM_CONTEXT.CURRENT_EVENT)93 else:94 context = {}95# no checkpoint specified, so start with an empty context96else:97 context = json.loads(args.initial_context or "{}", **json_loads_additional_kwargs())98 current_state = current_event = STATE.PSEUDO_INIT99# start things off100start_state_machine(args.machine_name,101 context,102 correlation_id=args.correlation_id,103 current_state=current_state,...
create_kinesis_stream.py
Source: create_kinesis_stream.py
1#!/usr/bin/env python2# Copyright 2016-2020 Workiva Inc.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15#16# create_kinesis_stream.py17#18# Script that create a Kinesis Stream.19# system imports20import argparse21import logging22import sys23# library imports24# application imports25from aws_lambda_fsm.aws import get_connection26from aws_lambda_fsm.aws import get_arn_from_arn_string27from aws_lambda_fsm.aws import validate_config28from aws_lambda_fsm.constants import AWS29import settings30# setup the command line args31parser = argparse.ArgumentParser(description='Creates AWS Kinesis streams.')32parser.add_argument('--kinesis_stream_arn', default='PRIMARY_STREAM_SOURCE')33parser.add_argument('--kinesis_num_shards', type=int, default=10)34parser.add_argument('--log_level', default='INFO')35parser.add_argument('--boto_log_level', default='INFO')36args = parser.parse_args()37logging.basicConfig(38 format='[%(levelname)s] %(asctime)-15s %(message)s',39 level=int(args.log_level) if args.log_level.isdigit() else args.log_level,40 datefmt='%Y-%m-%d %H:%M:%S'41)42logging.getLogger('boto3').setLevel(args.boto_log_level)43logging.getLogger('botocore').setLevel(args.boto_log_level)44validate_config()45# setup connections to AWS46kinesis_stream_arn = getattr(settings, args.kinesis_stream_arn)47logging.info('Kinesis stream ARN: %s', kinesis_stream_arn)48logging.info('Kinesis endpoint: %s', settings.ENDPOINTS.get(AWS.KINESIS))49if get_arn_from_arn_string(kinesis_stream_arn).service != AWS.KINESIS:50 logging.fatal("%s is not a Kinesis ARN", kinesis_stream_arn)51 sys.exit(1)52kinesis_conn = get_connection(kinesis_stream_arn, disable_chaos=True)53kinesis_stream = get_arn_from_arn_string(kinesis_stream_arn).slash_resource()54logging.info('Kinesis stream: %s', kinesis_stream)55# configure the stream56response = kinesis_conn.create_stream(57 StreamName=kinesis_stream,58 ShardCount=args.kinesis_num_shards59)...
stream_initial copy.py
Source: stream_initial copy.py
1import boto32import botostubs3from botocore.exceptions import ClientError4pinpoint_client = boto3.client('pinpoint') # type: botostubs.pinpoint5sts_client = boto3.client("sts")6account_id = sts_client.get_caller_identity()["Account"]7PINPOINT_ID = 'd5ab8b3e1788464cbc67948ab042fbcc'8KINESIS_STREAM_ARN = ''9EVENT_STREAM_IAM_ARN = ''10print('KINESIS_STREAM_ARN: ' + KINESIS_STREAM_ARN)11print('EVENT_STREAM_IAM_ARN ARN: ' + EVENT_STREAM_IAM_ARN)12print('Pinpoint event stream: creating')13try:14 response = pinpoint_client.put_event_stream(15 ApplicationId=PINPOINT_ID,16 WriteEventStream={17 'DestinationStreamArn': KINESIS_STREAM_ARN,18 'RoleArn': EVENT_STREAM_IAM_ARN19 }20 )21except ClientError as error:...
Check out the latest blogs from LambdaTest on this topic:
The fact is not alien to us anymore that cross browser testing is imperative to enhance your application’s user experience. Enhanced knowledge of popular and highly acclaimed testing frameworks goes a long way in developing a new app. It holds more significance if you are a full-stack developer or expert programmer.
QA testers have a unique role and responsibility to serve the customer. Serving the customer in software testing means protecting customers from application defects, failures, and perceived failures from missing or misunderstood requirements. Testing for known requirements based on documentation or discussion is the core of the testing profession. One unique way QA testers can both differentiate themselves and be innovative occurs when senseshaping is used to improve the application user experience.
Having a good web design can empower business and make your brand stand out. According to a survey by Top Design Firms, 50% of users believe that website design is crucial to an organization’s overall brand. Therefore, businesses should prioritize website design to meet customer expectations and build their brand identity. Your website is the face of your business, so it’s important that it’s updated regularly as per the current web design trends.
Enterprise resource planning (ERP) is a form of business process management software—typically a suite of integrated applications—that assists a company in managing its operations, interpreting data, and automating various back-office processes. The introduction of a new ERP system is analogous to the introduction of a new product into the market. If the product is not handled appropriately, it will fail, resulting in significant losses for the business. Most significantly, the employees’ time, effort, and morale would suffer as a result of the procedure.
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!