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

How to use events_rule_arn method in localstack

Best Python code snippet using localstack_python

events.py

Source:events.py

...62    def cloudformation_type():63        return "AWS::Events::Rule"64    def get_cfn_attribute(self, attribute_name):65        if attribute_name == "Arn":66            return self.params.get("Arn") or aws_stack.events_rule_arn(self.params.get("Name"))67        return super(EventsRule, self).get_cfn_attribute(attribute_name)68    def get_physical_resource_id(self, attribute=None, **kwargs):69        return self.props.get("Name")70    def fetch_state(self, stack_name, resources):71        rule_name = self.resolve_refs_recursively(stack_name, self.props.get("Name"), resources)72        result = aws_stack.connect_to_service("events").describe_rule(Name=rule_name) or {}73        return result if result.get("Name") else None74    @staticmethod75    def add_defaults(resource, stack_name: str):76        role_name = resource.get("Properties", {}).get("Name")77        if not role_name:78            resource["Properties"]["Name"] = generate_default_name(79                stack_name, resource["LogicalResourceId"]80            )...

lambda_service.py

Source:lambda_service.py

1from .base_service import BaseService2import zipfile3import os4import configparser5class Lambda_service(BaseService):6    """7    Lambda class, to help with Lambda related operations8    """9    def __init__(self, service_name='lambda', app_config="app_config.ini"):10        """11        Initiate AWS service class12        :param service_name: AWS service name13        :param app_config: App configuration file name14        """15        BaseService.__init__(self, service_name, app_config)16    def create_lambda(self, iam_role_arn: str, kms_key_arn: str, environment: str, tags: dict):17        """18        Create a lambda function19        :param iam_role_arn: IAM role ARN that needs to be attached to lambda20        :param kms_key_arn: KSM key ARN21        :param environment: Preferred running environment, such as Python22        :param tags: tags23        :return: Response from create lambda request24        """25        # zip dependencies26        def zip_lib(libraries, zf):27            # zf is zipfile handle28            lib_path = os.path.join(self.cwd, 'venv\Lib\site-packages')29            for lib in libraries:30                lib_full_path = f'{lib_path}\\{lib}'31                if os.path.isfile(lib_full_path):32                    zf.write(lib_full_path, os.path.basename(lib_full_path))33                elif os.path.isdir(lib_full_path):34                    for root, dirs, files in os.walk(lib_full_path):35                        for file in files:36                            zf.write(os.path.join(root, file), f'{lib}\\{file}')37        # Get current working directory38        cwd = self.cwd39        project_name = self.project_name40        lambda_file = os.path.join(cwd, f'{project_name}.py')41        setup_config = configparser.ConfigParser()42        setup_config.read(self.config_file)43        libs = list(dict(setup_config.items('lib'))['libs'].split(","))44        zip_file = os.path.join(cwd, f'{self.config["function_name"]}.zip')45        # Create lambda package46        zf = zipfile.ZipFile(zip_file, 'w', zipfile.ZIP_DEFLATED)47        try:48            print(f'Creating {self.config["function_name"]}.zip...')49            zf.write(lambda_file, os.path.basename(lambda_file))50            zip_lib(libs, zf)51        except Exception as e:52            self.logger.exception("")53            raise e54        else:55            print(f'{self.config["function_name"]}.zip is created')56        finally:57            print('Closing zipfile...')58            null = zf.close()59        # Create Lambda functions60        try:61            response = self.client.create_function(62                FunctionName=self.config['function_name'],63                Runtime=self.config['runtime'],64                Role=iam_role_arn,65                Handler=f'{self.config["function_name"]}.lambda_handler',66                Code={67                    'ZipFile': open(zip_file, 'rb').read()68                },69                Description=self.config['description'],70                Timeout=60,71                MemorySize=128,72                Publish=False,73                Environment=environment,74                KMSKeyArn=kms_key_arn,75                Tags=tags76            )77        except Exception as e:78            self.logger.exception("")79            raise e80        else:81            return response82    def add_trigger(self, events_rule_arn: str):83        """84        Create a trigger for lambda job85        :param events_rule_arn: EventsBridge Rule ARN86        :return: Response from lambda add_permission request87        """88        response = self.client.add_permission(89            FunctionName=self.config['function_name'],90            StatementId=self.config['statement_id'],91            Action="lambda:InvokeFunction",92            Principal="events.amazonaws.com",93            SourceArn=events_rule_arn94        )95        return response96    def delete_lambda(self):97        """98        Delete lambda function99        :return: Response from lambda delete request100        """101        try:102            response = self.client.delete_function(FunctionName=self.config['function_name'])103        except Exception as e:104            self.logger.exception("")105            raise e106        else:107            print(f'Lambda function {self.config["function_name"]} has been removed')...

service_models.py

Source:service_models.py

...17        return attr18class EventsRule(BaseModel):19    def get_cfn_attribute(self, attribute_name):20        if attribute_name == 'Arn':21            return self.params.get('Arn') or aws_stack.events_rule_arn(self.params.get('Name'))22        return super(EventsRule, self).get_cfn_attribute(attribute_name)23class LogsLogGroup(BaseModel):24    def get_cfn_attribute(self, attribute_name):25        if attribute_name == 'Arn':26            return self.params.get('Arn') or aws_stack.log_group_arn(self.params.get('LogGroupName'))27        return super(LogsLogGroup, self).get_cfn_attribute(attribute_name)28class CloudFormationStack(BaseModel):29    pass30class ElasticsearchDomain(BaseModel):31    pass32class FirehoseDeliveryStream(BaseModel):33    pass34class GatewayResponse(BaseModel):35    pass...

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.