How to use get_lambda_event_filters_for_arn method in localstack

Best Python code snippet using localstack_python

stream_event_source_listener.py

Source:stream_event_source_listener.py

...168            records_response = stream_client.get_records(169                ShardIterator=shard_iterator, Limit=batch_size170            )171            records = records_response.get("Records")172            event_filter_criterias = get_lambda_event_filters_for_arn(function_arn, stream_arn)173            if len(event_filter_criterias) > 0:174                records = filter_stream_records(records, event_filter_criterias)175            should_get_next_batch = True176            if records:177                payload = self._create_lambda_event_payload(stream_arn, records, shard_id=shard_id)178                is_invocation_successful, status_code = self._invoke_lambda(179                    function_arn, payload, lock_discriminator, parallelization_factor180                )181                if is_invocation_successful:182                    should_get_next_batch = True183                else:184                    num_invocation_failures += 1185                    if num_invocation_failures >= max_num_retries:186                        should_get_next_batch = True...

lambda_utils.py

Source:lambda_utils.py

...331            return False332    # needs to contain on what to filter (some list with citerias)333    # https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-syntax334    return True335def get_lambda_event_filters_for_arn(lambda_arn: str, event_arn: str) -> List[Dict]:336    # late import to avoid circular import337    from localstack.services.awslambda.lambda_api import LambdaRegion338    region_name = lambda_arn.split(":")[3]339    region = LambdaRegion.get(region_name)340    event_filter_criterias = [341        event_source_mapping.get("FilterCriteria")342        for event_source_mapping in region.event_source_mappings343        if event_source_mapping.get("FunctionArn") == lambda_arn344        and event_source_mapping.get("EventSourceArn") == event_arn345        and event_source_mapping.get("FilterCriteria") is not None346    ]...

sqs_event_source_listener.py

Source:sqs_event_source_listener.py

...163            }164            if md5OfMessageAttributes := msg.get("MD5OfMessageAttributes"):165                record["md5OfMessageAttributes"] = md5OfMessageAttributes166            records.append(record)167        event_filter_criterias = get_lambda_event_filters_for_arn(lambda_arn, queue_arn)168        if len(event_filter_criterias) > 0:169            # convert to json for filtering170            for record in records:171                try:172                    record["body"] = json.loads(record["body"])173                except json.JSONDecodeError:174                    LOG.warning(175                        f"Unable to convert record '{record['body']}' to json... Record might be dropped."176                    )177            records = filter_stream_records(records, event_filter_criterias)178            # convert them back179            for record in records:180                record["body"] = (181                    json.dumps(record["body"])...

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.