How to use get_kinesis_stream_name method in localstack

Best Python code snippet using localstack_python

34079_dynamodbstreams_api.py

Source:34079_dynamodbstreams_api.py

...10ACTION_HEADER_PREFIX = 'DynamoDBStreams_20120810'11def add_dynamodb_stream(table_name, view_type='NEW_AND_OLD_IMAGES', enabled=True):12    if enabled:13        # create kinesis stream as a backend14        stream_name = get_kinesis_stream_name(table_name)15        aws_stack.create_kinesis_stream(stream_name)16        stream = {17            'StreamArn': aws_stack.dynamodb_stream_arn(table_name=table_name),18            'TableName': table_name,19            'StreamLabel': 'TODO',20            'StreamStatus': 'ENABLED',21            'KeySchema': [],22            'Shards': []23        }24        table_arn = aws_stack.dynamodb_table_arn(table_name)25        DDB_STREAMS[table_arn] = stream26def forward_events(records):27    kinesis = aws_stack.connect_to_service('kinesis')28    for record in records:29        table_arn = record['eventSourceARN']30        stream = DDB_STREAMS.get(table_arn)31        if stream:32            table_name = table_name_from_stream_arn(stream['StreamArn'])33            stream_name = get_kinesis_stream_name(table_name)34            kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')35@app.route('/', methods=['POST'])36def post_request():37    action = request.headers.get('x-amz-target')38    data = json.loads(to_str(request.data))39    result = {}40    kinesis = aws_stack.connect_to_service('kinesis')41    if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:42        result = {43            'Streams': list(DDB_STREAMS.values()),44            'LastEvaluatedStreamArn': 'TODO'45        }46    elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:47        for stream in DDB_STREAMS.values():48            if stream['StreamArn'] == data['StreamArn']:49                result = {50                    'StreamDescription': stream51                }52                # get stream details53                dynamodb = aws_stack.connect_to_service('dynamodb')54                table_name = table_name_from_stream_arn(stream['StreamArn'])55                stream_name = get_kinesis_stream_name(table_name)56                stream_details = kinesis.describe_stream(StreamName=stream_name)57                table_details = dynamodb.describe_table(TableName=table_name)58                stream['KeySchema'] = table_details['Table']['KeySchema']59                stream['Shards'] = stream_details['StreamDescription']['Shards']60                break61        if not result:62            return error_response('Requested resource not found', error_type='ResourceNotFoundException')63    elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:64        # forward request to Kinesis API65        stream_name = stream_name_from_stream_arn(data['StreamArn'])66        result = kinesis.get_shard_iterator(StreamName=stream_name,67            ShardId=data['ShardId'], ShardIteratorType=data['ShardIteratorType'])68    elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:69        kinesis_records = kinesis.get_records(**data)70        result = {'Records': []}71        for record in kinesis_records['Records']:72            result['Records'].append(json.loads(to_str(record['Data'])))73    else:74        print('WARNING: Unknown operation "%s"' % action)75    return jsonify(result)76# -----------------77# HELPER FUNCTIONS78# -----------------79def error_response(message=None, error_type=None, code=400):80    if not message:81        message = 'Unknown error'82    if not error_type:83        error_type = 'UnknownError'84    if 'com.amazonaws.dynamodb' not in error_type:85        error_type = 'com.amazonaws.dynamodb.v20120810#%s' % error_type86    content = {87        'message': message,88        '__type': error_type89    }90    return make_response(jsonify(content), code)91def get_kinesis_stream_name(table_name):92    return DDB_KINESIS_STREAM_NAME_PREFIX + table_name93def table_name_from_stream_arn(stream_arn):94    return stream_arn.split(':table/')[1].split('/')[0]95def stream_name_from_stream_arn(stream_arn):96    table_name = table_name_from_stream_arn(stream_arn)97    return get_kinesis_stream_name(table_name)98def serve(port, quiet=True):...

dynamodbstreams_api.py

Source:dynamodbstreams_api.py

...14ACTION_HEADER_PREFIX = 'DynamoDBStreams_20120810'15def add_dynamodb_stream(table_name, view_type='NEW_AND_OLD_IMAGES', enabled=True):16    if enabled:17        # create kinesis stream as a backend18        stream_name = get_kinesis_stream_name(table_name)19        aws_stack.create_kinesis_stream(stream_name)20        stream = {21            'StreamArn': aws_stack.dynamodb_stream_arn(table_name=table_name),22            'TableName': table_name,23            'StreamLabel': 'TODO',24            'StreamStatus': 'ENABLED',25            'KeySchema': [],26            'Shards': []27        }28        table_arn = aws_stack.dynamodb_table_arn(table_name)29        DDB_STREAMS[table_arn] = stream30def forward_events(records):31    kinesis = aws_stack.connect_to_service('kinesis')32    for record in records:33        table_arn = record['eventSourceARN']34        stream = DDB_STREAMS.get(table_arn)35        if stream:36            table_name = table_name_from_stream_arn(stream['StreamArn'])37            stream_name = get_kinesis_stream_name(table_name)38            kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')39@app.route('/', methods=['POST'])40def post_request():41    action = request.headers.get('x-amz-target')42    data = json.loads(to_str(request.data))43    result = None44    kinesis = aws_stack.connect_to_service('kinesis')45    if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:46        result = {47            'Streams': list(DDB_STREAMS.values()),48            'LastEvaluatedStreamArn': 'TODO'49        }50    elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:51        for stream in DDB_STREAMS.values():52            if stream['StreamArn'] == data['StreamArn']:53                result = {54                    'StreamDescription': stream55                }56                # get stream details57                dynamodb = aws_stack.connect_to_service('dynamodb')58                table_name = table_name_from_stream_arn(stream['StreamArn'])59                stream_name = get_kinesis_stream_name(table_name)60                stream_details = kinesis.describe_stream(StreamName=stream_name)61                table_details = dynamodb.describe_table(TableName=table_name)62                stream['KeySchema'] = table_details['Table']['KeySchema']63                stream['Shards'] = stream_details['StreamDescription']['Shards']64                break65    elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:66        # forward request to Kinesis API67        stream_name = stream_name_from_stream_arn(data['StreamArn'])68        result = kinesis.get_shard_iterator(StreamName=stream_name,69            ShardId=data['ShardId'], ShardIteratorType=data['ShardIteratorType'])70    elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:71        kinesis_records = kinesis.get_records(**data)72        result = {'Records': []}73        for record in kinesis_records['Records']:74            result['Records'].append(json.loads(to_str(record['Data'])))75    else:76        print('WARNING: Unknown operation "%s"' % action)77    return jsonify(result)78# -----------------79# HELPER FUNCTIONS80# -----------------81def get_kinesis_stream_name(table_name):82    return DDB_KINESIS_STREAM_NAME_PREFIX + table_name83def table_name_from_stream_arn(stream_arn):84    return stream_arn.split(':table/')[1].split('/')[0]85def stream_name_from_stream_arn(stream_arn):86    table_name = table_name_from_stream_arn(stream_arn)87    return get_kinesis_stream_name(table_name)88def serve(port, quiet=True):89    if quiet:90        log = logging.getLogger('werkzeug')91        log.setLevel(logging.ERROR)92    ssl_context = GenericProxy.get_flask_ssl_context()93    app.run(port=int(port), threaded=True, host='0.0.0.0', ssl_context=ssl_context)94if __name__ == '__main__':95    port = DEFAULT_PORT_DYNAMODBSTREAMS96    print("Starting server on port %s" % port)...

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.