Best Python code snippet using localstack_python
kinesis_listener.py
Source:kinesis_listener.py
...70 return True71 def return_response(self, method, path, data, headers, response):72 action = headers.get('X-Amz-Target')73 data = self.decode_content(data or '{}')74 response._content = self.replace_in_encoded(response.content or '')75 records = []76 if action in (ACTION_CREATE_STREAM, ACTION_DELETE_STREAM):77 event_type = (event_publisher.EVENT_KINESIS_CREATE_STREAM if action == ACTION_CREATE_STREAM78 else event_publisher.EVENT_KINESIS_DELETE_STREAM)79 payload = {'n': event_publisher.get_hash(data.get('StreamName'))}80 if action == ACTION_CREATE_STREAM:81 payload['s'] = data.get('ShardCount')82 event_publisher.fire_event(event_type, payload=payload)83 elif action == ACTION_PUT_RECORD:84 response_body = self.decode_content(response.content)85 # Note: avoid adding 'encryptionType':'NONE' in the event_record, as this breaks .NET Lambdas86 event_record = {87 'approximateArrivalTimestamp': epoch_timestamp(),88 'data': data['Data'],89 'partitionKey': data['PartitionKey'],90 'sequenceNumber': response_body.get('SequenceNumber')91 }92 event_records = [event_record]93 stream_name = data['StreamName']94 lambda_api.process_kinesis_records(event_records, stream_name)95 elif action == ACTION_PUT_RECORDS:96 event_records = []97 response_body = self.decode_content(response.content)98 if 'Records' in response_body:99 response_records = response_body['Records']100 records = data['Records']101 for index in range(0, len(records)):102 record = records[index]103 # Note: avoid adding 'encryptionType':'NONE' in the event_record, as this breaks .NET Lambdas104 event_record = {105 'approximateArrivalTimestamp': epoch_timestamp(),106 'data': record['Data'],107 'partitionKey': record['PartitionKey'],108 'sequenceNumber': response_records[index].get('SequenceNumber')109 }110 event_records.append(event_record)111 stream_name = data['StreamName']112 lambda_api.process_kinesis_records(event_records, stream_name)113 elif action == ACTION_UPDATE_SHARD_COUNT:114 # Currently kinesalite, which backs the Kinesis implementation for localstack, does115 # not support UpdateShardCount:116 # https://github.com/mhart/kinesalite/issues/61117 #118 # [Terraform](https://www.terraform.io) makes the call to UpdateShardCount when it119 # applies Kinesis resources. A Terraform run fails when this is not present.120 #121 # The code that follows just returns a successful response, bypassing the 400122 # response that kinesalite returns.123 #124 response = Response()125 response.status_code = 200126 content = {127 'CurrentShardCount': 1,128 'StreamName': data['StreamName'],129 'TargetShardCount': data['TargetShardCount']130 }131 response.encoding = 'UTF-8'132 response._content = json.dumps(content)133 return response134 elif action == ACTION_GET_RECORDS:135 sdk_v2 = self.sdk_is_v2(headers.get('User-Agent', '').split(' ')[0])136 results, encoding_type = self.decode_content(response.content, True)137 for record in results['Records']:138 if sdk_v2:139 record['ApproximateArrivalTimestamp'] = int(record['ApproximateArrivalTimestamp'] * 1000)140 if not isinstance(record['Data'], str):141 record['Data'] = base64.encodebytes(bytearray(record['Data']['data']))142 if encoding_type == APPLICATION_CBOR:143 response._content = cbor2.dumps(results)144 else:145 response._content = json.dumps(results)146 return response147 def sdk_is_v2(self, user_agent):148 if re.search(r'\/2.\d+.\d+', user_agent):149 return True150 return False151 def replace_in_encoded(self, data):152 if not data:153 return ''154 decoded, type_encoding = self.decode_content(data, True)155 if type_encoding == APPLICATION_JSON:156 return re.sub(r'arn:aws:kinesis:[^:]+:', 'arn:aws:kinesis:%s:' % aws_stack.get_region(),157 to_str(data))158 if type_encoding == APPLICATION_CBOR:159 replaced = re.sub(r'arn:aws:kinesis:[^:]+:', 'arn:aws:kinesis:%s:' % aws_stack.get_region(),160 json.dumps(decoded))161 return cbor2.dumps(json.loads(replaced))162 def decode_content(self, data, describe=False):163 content_type = ''164 try:165 decoded = json.loads(to_str(data))...
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!!