How to use list_partner_event_sources method in localstack

Best Python code snippet using localstack_python

client.pyi

Source:client.pyi Github

copy

Full Screen

...430 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.440 [Show boto3 documentation](https://boto3.amazonaws.com/v1/documentation/api/1.24.58/reference/services/events.html#EventBridge.Client.list_partner_event_sources)441 [Show boto3-stubs documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_events/client.html#list_partner_event_sources)442 """443 def list_replays(444 self,445 *,446 NamePrefix: str = None,447 State: ReplayStateType = None,448 EventSourceArn: str = None,...

Full Screen

Full Screen

aws_events_info.py

Source:aws_events_info.py Github

copy

Full Screen

...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'):184 paginator = client.get_paginator('list_replays')185 return paginator.paginate(186 NamePrefix=module.params['name_prefix'],187 ), True188 else:189 return client.list_replays(190 NamePrefix=module.params['name_prefix'],191 ), False192 elif module.params['list_rules']:193 if client.can_paginate('list_rules'):...

Full Screen

Full Screen

events_client.py

Source:events_client.py Github

copy

Full Screen

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)...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful