How to use list_partner_event_source_accounts method in localstack

Best Python code snippet using localstack_python

client.pyi

Source:client.pyi

...421        your Amazon Web Services account.422        [Show boto3 documentation](https://boto3.amazonaws.com/v1/documentation/api/1.24.58/reference/services/events.html#EventBridge.Client.list_event_sources)423        [Show boto3-stubs documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_events/client.html#list_event_sources)424        """425    def list_partner_event_source_accounts(426        self, *, EventSourceName: str, NextToken: str = None, Limit: int = None427    ) -> ListPartnerEventSourceAccountsResponseTypeDef:428        """429        An SaaS partner can use this operation to display the Amazon Web Services430        account ID that a particular partner event source name is associated with.431        [Show boto3 documentation](https://boto3.amazonaws.com/v1/documentation/api/1.24.58/reference/services/events.html#EventBridge.Client.list_partner_event_source_accounts)432        [Show boto3-stubs documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_events/client.html#list_partner_event_source_accounts)433        """434    def list_partner_event_sources(435        self, *, NamePrefix: str, NextToken: str = None, Limit: int = None436    ) -> ListPartnerEventSourcesResponseTypeDef:437        """438        An SaaS partner can use this operation to list all the partner event source439        names that they have created....

aws_events_info.py

Source:aws_events_info.py

...165                return paginator.paginate(166                    EventSourceName=module.params['event_source_name'],167                ), True168            else:169                return client.list_partner_event_source_accounts(170                    EventSourceName=module.params['event_source_name'],171                ), False172        elif module.params['list_partner_event_sources']:173            if client.can_paginate('list_partner_event_sources'):174                paginator = client.get_paginator('list_partner_event_sources')175                return paginator.paginate(176                    NamePrefix=module.params['name_prefix'],177                ), True178            else:179                return client.list_partner_event_sources(180                    NamePrefix=module.params['name_prefix'],181                ), False182        elif module.params['list_replays']:183            if client.can_paginate('list_replays'):...

events_client.py

Source:events_client.py

1"""2AWS client to handle service API requests.3"""4from horey.aws_api.aws_clients.boto3_client import Boto3Client5from horey.aws_api.aws_services_entities.event_bridge_rule import EventBridgeRule6from horey.aws_api.aws_services_entities.event_bridge_target import EventBridgeTarget7from horey.aws_api.base_entities.aws_account import AWSAccount8from horey.h_logger import get_logger9logger = get_logger()10import pdb11class EventsClient(Boto3Client):12    """13    Client to handle specific aws service API calls.14    """15    def __init__(self):16        client_name = "events"17        super().__init__(client_name)18    def get_all_rules(self, region=None, full_information=True):19        """20        Get all rules in all regions.21        :return:22        """23        if region is not None:24            return self.get_region_rules(region, full_information=full_information)25        final_result = list()26        for region in AWSAccount.get_aws_account().regions.values():27            final_result += self.get_region_rules(region, full_information=full_information)28        return final_result29    def get_region_rules(self, region, full_information=True, custom_filter=None):30        final_result = list()31        AWSAccount.set_aws_region(region)32        for dict_src in self.execute(self.client.list_rules, "Rules", filters_req=custom_filter):33            obj = EventBridgeRule(dict_src)34            final_result.append(obj)35            if full_information:36                self.update_rule_targets(obj)37                self.update_rule_tags(obj)38                pass39        return final_result40    def update_rule_targets(self, rule):41        filters_req = {"Rule": rule.name}42        rule.targets = []43        for dict_src in self.execute(self.client.list_targets_by_rule, "Targets", filters_req=filters_req):44            obj = EventBridgeTarget(dict_src)45            rule.targets.append(obj)46    def update_rule_tags(self, rule):47        filters_req = {"ResourceARN": rule.arn}48        rule.tags = []49        for dict_src in self.execute(self.client.list_tags_for_resource, "Tags", filters_req=filters_req):50            rule.targets.append(dict_src)51    def get_samples(self):52        ret = list(self.execute(self.client.list_connections, None, raw_data=True))53        ret = list(self.execute(self.client.list_event_buses, None, raw_data=True))54        ret = list(self.execute(self.client.list_event_sources, None, raw_data=True))55        ret = list(self.execute(self.client.list_partner_event_source_accounts, None, raw_data=True))56        ret = list(self.execute(self.client.list_partner_event_sources, None, raw_data=True))57        ret = list(self.execute(self.client.list_replays, None, raw_data=True))58        ret = list(self.execute(self.client.list_rule_names_by_target, None, raw_data=True))59        ret = list(self.execute(self.client.list_targets_by_rule, None, raw_data=True, filters_req={"Rule":"initiator-production-InitiatorEventsRuleSchedule1-MWYJ6917GUN9"}))60    def update_rule_information(self, rule):61        region_rules = self.get_region_rules(rule.region, custom_filter={"NamePrefix": rule.name})62        if len(region_rules) == 1:63            rule.update_from_raw_response(region_rules[0].dict_src)64            return65        if len(region_rules) > 1:66            raise RuntimeError(region_rules)67    def provision_rule(self, rule: EventBridgeRule):68        self.update_rule_information(rule)69        AWSAccount.set_aws_region(rule.region)70        if rule.arn is None:71            response = self.provision_rule_raw(rule.generate_create_request())72            del response["ResponseMetadata"]73            rule.update_from_raw_response(response)74        put_targets_request = rule.generate_put_targets_request()75        if put_targets_request is not None:76            self.put_targets_raw(put_targets_request)77    def provision_rule_raw(self, request_dict):78        logger.info(f"Creating rule: {request_dict}")79        for response in self.execute(self.client.put_rule, None, raw_data=True,80                                     filters_req=request_dict):81            return response82    def put_targets_raw(self, request_dict):83        logger.info(f"Putting targets: {request_dict}")84        for response in self.execute(self.client.put_targets, None, raw_data=True, filters_req=request_dict):85            if response.get("FailedEntryCount") != 0:86                raise RuntimeError(response)...

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.