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

How to use get_event_source_mappings method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

...80        aliases.extend(page['Aliases'])81    return aliases82@timeit83@aws_handle_regions84def get_event_source_mappings(lambda_function: Dict, client: botocore.client.BaseClient) -> List[Any]:85    event_source_mappings: List[Any] = []86    paginator = client.get_paginator('list_event_source_mappings')87    for page in paginator.paginate(FunctionName=lambda_function['FunctionName']):88        event_source_mappings.extend(page['EventSourceMappings'])89    return event_source_mappings90@timeit91@aws_handle_regions92def get_lambda_function_details(93        boto3_session: boto3.session.Session, data: List[Dict], region: str,94) -> Generator[Any, Any, None]:95    client = boto3_session.client('lambda', region_name=region)96    for lambda_function in data:97        function_aliases = get_function_aliases(lambda_function, client)98        event_source_mappings = get_event_source_mappings(lambda_function, client)99        layers = lambda_function.get('Layers', [])100        yield lambda_function['FunctionArn'], function_aliases, event_source_mappings, layers101@timeit102def load_lambda_function_details(103        neo4j_session: neo4j.Session, lambda_function_details: List[Tuple[str, List[Dict], List[Dict], List[Dict]]],104        update_tag: int,105) -> None:106    lambda_aliases: List[Dict] = []107    lambda_event_source_mappings: List[Dict] = []108    lambda_layers: List[Dict] = []109    for function_arn, aliases, event_source_mappings, layers in lambda_function_details:110        if len(aliases) > 0:111            for alias in aliases:112                alias['FunctionArn'] = function_arn...

lambda_functions.py

Source:lambda_functions.py

...12            else self.client.list_functions(MaxItems=20),13            "NextMarker",14            "Functions",15        )16    def get_event_source_mappings(self):17        return self.get_paginated_results(18            lambda marker: self.client.list_event_source_mappings(19                Marker=marker, MaxItems=2020            )21            if marker22            else self.client.list_event_source_mappings(MaxItems=20),23            "NextMarker",24            "EventSourceMappings",25        )26class LambdaResultsParser:27    def create_function_nodes(self, items):28        function_nodes = []29        for item in items:30            function_nodes.append(self.create_function_node(item))31        return function_nodes32    @staticmethod33    def create_function_node(function):34        return models.Resource(35            name=function["FunctionName"],36            resource_type="Lambda-Function",37            id=function["FunctionArn"],38            service="AWS-Lambda",39            category="COMPUTE",40        )41    def create_event_source_links(self, items):42        event_source_links = []43        for event_source in items:44            event_source_links.append(self.create_event_source_link(event_source))45            event_source_links.extend(46                self.create_destination_config_links(event_source)47            )48        return event_source_links49    @staticmethod50    def create_event_source_link(event_source):51        return models.Link(52            source=event_source["EventSourceArn"],53            destination=event_source["FunctionArn"],54            link_type="",55        )56    def create_destination_config_links(self, item):57        config_links = []58        config = item.get("DestinationConfig", {})59        self._create_config_link(config, "OnSuccess", config_links, "")60        self._create_config_link(config, "OnFailure", config_links, "DLQ")61        return config_links62    @staticmethod63    def _create_config_link(item, config_name, config_links, link_name):64        config = item.get(config_name, {})65        if config:66            config_links.append(67                models.Link(68                    source=config["FunctionArn"],69                    destination=config["Destination"],70                    link_type=link_name,71                )72            )73def get(nodes, links, region):74    logging.info("Starting Lambda collection")75    collector = LambdaDataCollector(region)76    items = collector.get_functions()77    parser = LambdaResultsParser()78    nodes.extend(parser.create_function_nodes(items))79    event_sources = collector.get_event_source_mappings()80    links.extend(parser.create_event_source_links(event_sources))...

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.