Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_event_source_mapping method in localstack

Best Python code snippet using localstack_python

check_object.py

Source:check_object.py

...63                self.logger_handler.log_message(self.s3_handler.create_s3_notification('innowise', 's3-notification'))64            except:65                self.logger_handler.log_error('Notification are created.')66            try:67                if self.lambda_handler.get_event_source_mapping('loader_to_dynamo_for_stage_data', 'sqs', 'stage-files-queue')['EventSourceMappings'].__len__() <= 0:68                    self.logger_handler.log_message(self.lambda_handler.create_event_source_mapping('loader_to_dynamo_for_stage_data', 'sqs', 'stage-files-queue'))69                if self.lambda_handler.get_event_source_mapping('loader_to_dynamo_for_parse_data_type_one', 'sqs', 'parse-files-queue-type-one')['EventSourceMappings'].__len__() <= 0:70                    self.logger_handler.log_message(self.lambda_handler.create_event_source_mapping('loader_to_dynamo_for_parse_data_type_one', 'sqs', 'parse-files-queue-type-one'))71                if self.lambda_handler.get_event_source_mapping('loader_to_dynamo_for_parse_data_type_two', 'sqs', 'parse-files-queue-type-two')['EventSourceMappings'].__len__() <= 0:72                    self.logger_handler.log_message(self.lambda_handler.create_event_source_mapping('loader_to_dynamo_for_parse_data_type_two', 'sqs', 'parse-files-queue-type-two'))73                74                if self.lambda_handler.get_event_source_mapping('loader_to_dynamo_for_metrics', 'sqs', 'metrics-queue')['EventSourceMappings'].__len__() <= 0:75                    self.logger_handler.log_message(self.lambda_handler.create_event_source_mapping('loader_to_dynamo_for_metrics', 'sqs', 'metrics-queue'))76            except:77                self.logger_handler.log_error('Event sourse mapping are created.')78            try:79                self.logger_handler.log_message(self.dynamodb_handler.create_table(80                    table_name='stage_data',81                    key_schema=[82                        {83                            'AttributeName': 'file_name',84                            'KeyType': 'HASH'85                        },86                        {87                            'AttributeName': 'departure_name',88                            'KeyType': 'RANGE'...

helper.py

Source:helper.py

...44            logging.debug(e)45            return None46    def function_exists(self, function_name) -> bool:47        return self.get_function(function_name) is not None48    def get_event_source_mapping(self, esm_uuid: str) -> dict:49        try:50            resp = self.lambda_client.get_event_source_mapping(51                UUID=esm_uuid,52            )53            return resp54        except Exception as e:55            logging.debug(e)56            return None57    def event_source_mapping_exists(self, esm_uuid: str) -> bool:58        return self.get_event_source_mapping(esm_uuid) is not None59    def get_code_signing_config(self, code_signing_config_arn: str) -> dict:60        try:61            resp = self.lambda_client.get_code_signing_config(62                CodeSigningConfigArn=code_signing_config_arn,63            )64            return resp["CodeSigningConfig"]65        except Exception as e:66            logging.debug(e)67            return None68    def code_signing_config_exists(self, code_signing_config_arn: str) -> bool:69        return self.get_code_signing_config(code_signing_config_arn) is not None70    def get_alias(self, alias_name: str, function_name: str) -> dict:71        try:72            resp = self.lambda_client.get_alias(...

kafka.py

Source:kafka.py

...16                try:17                    print("EventSourceMapping recreation")18                    client.delete_event_source_mapping(UUID=physicalResourceId)19                    while True:20                        client.get_event_source_mapping(UUID=physicalResourceId)21                        time.sleep(10)22                except client.exceptions.ResourceNotFoundException:23                    pass24            response = client.create_event_source_mapping(25                FunctionName=event["ResourceProperties"]["Function"],26                BatchSize=int(event["ResourceProperties"]["BatchSize"]),27                StartingPosition=event["ResourceProperties"]["StartingPosition"],28                Topics=[29                    event["ResourceProperties"]["Topic"]30                ],31                SelfManagedEventSource={32                    "Endpoints": {33                        "KAFKA_BOOTSTRAP_SERVERS": event["ResourceProperties"]["Brokers"]34                    }35                },36                SourceAccessConfigurations=list([37                    {38                        "Type": "VPC_SUBNET",39                        "URI": "subnet:" + subnetId40                    } for subnetId in event["ResourceProperties"]["SubnetIds"]41                ]) + list([42                    {43                        "Type": "VPC_SECURITY_GROUP",44                        "URI": "security_group:" + securityGroupId45                    } for securityGroupId in event["ResourceProperties"]["SecurityGroupIds"]46                ])47            )48            physicalResourceId = response["UUID"]49            while True:50                response = client.get_event_source_mapping(UUID=physicalResourceId)51                if response["State"] in ["Enabled", "Disabled"]:52                    break53                time.sleep(10)54            print("EventSourceMapping successfully created")55        elif event["RequestType"] == "Delete":56            try:57                client.delete_event_source_mapping(UUID=physicalResourceId)58            except client.exceptions.ResourceNotFoundException:59                pass60            print("EventSourceMapping successfully deleted")61    except Exception as exc:62        print("Failed to process:", exc)63        responseStatus = "FAILED"64    finally:...

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.