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

How to use describe_stream method in localstack

Best Python code snippet using localstack_python

aws_dynamodbstreams_info.py

Source:aws_dynamodbstreams_info.py

...131                return client.list_streams(132                    TableName=module.params['name'],133                ), False134        elif module.params['describe_stream']:135            return client.describe_stream(136                StreamArn=module.params['stream_arn'],137            ), False138        elif module.params['get_shard_iterator']:139            return client.describe_stream(140                StreamArn=module.params['stream_arn'],141                ShardId=module.params['shard_id'],142                ShardIteratorType=module.params['shard_iterator_type'],143                SequenceNumber=module.params['sequence_number'],144            ), False145        elif module.params['get_records']:146            if client.can_paginate('get_records'):147                paginator = client.get_paginator('get_records')148                return paginator.paginate(149                    ShardIterator=module.params['shard_iterator'],150                ), True151            else:152                return client.get_records(153                    ShardIterator=module.params['shard_iterator'],...

kinesis_lab.py

Source:kinesis_lab.py

...5cli_ken = session.client("kinesis")6def create_stream(name,shards):7    res_create_stream = cli_ken.create_stream(StreamName=name, ShardCount = shards)8    return res_create_stream9def describe_stream(stream_name):10    res_desc_stream = cli_ken.describe_stream(StreamName=stream_name)11    return res_desc_stream12def list_stream(limit):13    res_list_stream = cli_ken.list_streams(Limit = limit)14    return res_list_stream15def get_kinesis_shards(stream_name, shard_id):16    res_shard_iter = cli_ken.get_shard_iterator(StreamName = stream_name, ShardId = shard_id, ShardIteratorType = 'LATEST')17    return res_shard_iter18def list_shards(stream_name):19    res_shard_lst = cli_ken.list_shards(StreamName = stream_name)20    lst_shards = [shard_id['ShardId'] for shard_id in res_shard_lst['Shards']]     21    return lst_shards22def get_record(shard_key):23    res_get_rec =  cli_ken.get_records(ShardIterator = shard_key, Limit = 10)24    return res_get_rec['Records']25def put_record(stream_name, data, key):26    res_put_rec = cli_ken.put_record(StreamName = stream_name, Data = data, PartitionKey = key)27    print ("Results from PUT: ShardId = {} and Sequence Number = {}".format (res_put_rec ['ShardId'], res_put_rec['SequenceNumber']))28    return [res_put_rec ['ShardId'], res_put_rec['SequenceNumber']]29    30def del_stream(stream_name):31    res_stream_del =cli_ken.delete_stream(StreamName=stream_name)32    return res_stream_del33if __name__ == '__main__':34    35    found=False36    stream_name = "Aji_Stream"37    shards = 138    #Create a stream   39    create_stream(stream_name,shards)    40    #Descrive a stream to get the Stream Status41    stream_status = describe_stream(stream_name)['StreamDescription'] ['StreamStatus']42    #Loop in till the stream is active43    while stream_status != "ACTIVE":44        print("Waiting for the stream to be active....")45        t.sleep(10)46        if describe_stream(stream_name)['StreamDescription'] ['StreamStatus'] == "ACTIVE":47            found=True48            break49    #If the Stream is Active.. Get the Shard ID50    if found==True:        51        shard_id  = describe_stream(stream_name)['StreamDescription'] ['Shards'][0]['ShardId']52    #If it gets a Shard ID.. Create the Shard Iterator53    if shard_id != "":    54        shard_key = get_kinesis_shards(stream_name, shard_id)['ShardIterator']55    56    #If the Get Data returns null.. Input a record to the stream    57    if len(get_record(shard_key)) == 0:58        shard_seq = put_record(stream_name,"Aji_Test_Data", "1")59    60    #Get the result of the data put inside the stream in the earlier command        61    if len(get_record(shard_key)) != 0:            62        result = [str(val['Data']) for val in get_record(shard_key)]63        print(result)64    # Delete the Stream        65    del_stream(stream_name)    66  67    #Verify whether the stream is deleated.68    stream_status = describe_stream(stream_name)['StreamDescription'] ['StreamStatus']        69    while stream_status != 'ACTIVE':70        try:71            print("Waiting for the stream to be deleted....")72            t.sleep(10)         73            stream_status = describe_stream(stream_name)['StreamDescription'] ['StreamStatus']74            if stream_status != 'DELETING':            75                break76        except ResourceNotFoundException:77            print("Stream Deleted.....")...

kinesis_utils.py

Source:kinesis_utils.py

...17    def __init__(self, stream_name, region='us-east-2'):18        self.stream_name         = stream_name19        self.region              = region20        self.kinesis_conn        = kinesis.connect_to_region(region)    # Connect to a Kinesis stream in a specific region21        self.description         = self.kinesis_conn.describe_stream(stream)22        self.shards              = self.get_shards()23        self.first_shard_id      = self.shards[0]["ShardId"]24        self.partition_key       = 025        26    def get_shards(self, description):27        """28        Parse shard information29        """30        return self.description['StreamDescription']['Shards']31    def get_partition_key(self):32        """33        Gets an increments the partition key34        """35        self.partition_key += 136        if self.partition_key >= len(self.shards):37            self.partition_key = 038            39        return self.partiton_key40        41    def put_record(self, record):42        """43        Put a record (<=1MB) on to a Kinesis stream44        """45        kinesis.put_record(self.stream_name, json.dumps(record), self.get_partition_key())46    def put_records(self, records):47        """48        Put records on to a Kinesis stream49        """50        data = []51        for record in records:52            data.append( { 'Data': json.dumps(record), 'PartitionKey': self.get_partition_key() } )53            54        kinesis.put_records(data, self.stream_name)55        56    def records_iterator(self, limit=100):57        """58        Read the latest batch of records from a Kinesis stream59        """60        shard_it = kinesis.get_shard_iterator(self.stream_name, self.first_shard_id, "LATEST")["ShardIterator"]61        while True:62            t0 = time.time()63            out = kinesis.get_records(shard_it, limit=limit)64            shard_it = out["NextShardIterator"]65            yield out66            67            t1 = time.time()68            if t1 - t0 < 0.2:69                time.sleep(0.2 - (t1 - t0))70        71################################################################################72# MAIN73if __name__ == '__main__':74    parser = argparser({'desc': "Helper functions for AWS Kinesis: kinesis_utils.py"})75    parser.add_argument('--describe_stream', help='Describe a specific Kinesis stream', required=False, type=str)76    args = parser.parse_args()   # Get inputs and options77    if args.describe_stream:78        describe_stream(args.describe_stream)79        80        ...

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.