Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use kinesis_stream_arn method in localstack

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

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.