How to use create_scheduled_action method in localstack

Best Python code snippet using localstack_python

repo.py

Source:repo.py

...141        schedule,142        override_config=None143    ):144        data = {"override_config": override_config}145        return self.create_scheduled_action(pulp, action='/sync/', schedule=schedule, data=data)146    def get_scheduled_sync(147        self,148        pulp,149        id150    ):151        return self.get_scheduled_action(pulp, action='/sync/', id=id)152    def list_scheduled_sync(153        self,154        pulp,155    ):156        return self.list_scheduled_action(pulp, action='/sync/')157    def augment_proxy(self, pulp, host, port, username=None, password=None):158        """159        set proxy_ options160        """161        importer_config = dict(host=host, port=port)162        if username is not None:163            importer_config['proxy_username'] = username164        if password is not None:165            importer_config['proxy_password'] = password166        with pulp.asserting(True):167            pulp.send(self.request('PUT', data=dict(importer_config=importer_config)))168        self.reload(pulp)169class Distributor(item.AssociatedItem):170    path = '/distributors/'171    relevant_data_keys = ['id', 'distributor_type_id', 'repo_id', 'config', 'last_publish', 'auto_publish']172    def schedule_publish(173        self,174        pulp,175        schedule,176        override_config=None177    ):178        data = {"override_config": override_config}179        return self.create_scheduled_action(pulp, action='/publish/', schedule=schedule, data=data)180    def get_scheduled_publish(181        self,182        pulp,183        id184    ):185        return self.get_scheduled_action(pulp, action='/publish/', id=id)186    def list_scheduled_publish(187        self,188        pulp,189    ):190        return self.list_scheduled_action(pulp, action='/publish/')191    def content_url(self, pulp, path='/'):192        '''assemblel url of the content being served'''193        return path_join(pulp.url, content_path,...

consumer_class.py

Source:consumer_class.py

...82        data = {83            "units": [{"unit_key": unit_key, "type_id": type_id}],84            "options": options85        }86        return self.create_scheduled_action(pulp, action='/content/install/', schedule=schedule, data=data)87    def schedule_update(88        self,89        pulp,90        schedule,91        type_id,92        unit_key,93        options=None94    ):95        data = {96            "units": [{"unit_key": unit_key, "type_id": type_id}],97            "options": options98        }99        return self.create_scheduled_action(pulp, action='/content/update/', schedule=schedule, data=data)100    def applicability(101        self,102        pulp,103        path='/actions/content/regenerate_applicability/'104    ):105        return pulp.send(self.request('POST', path=path))106    def schedule_uninstall(107        self,108        pulp,109        schedule,110        type_id,111        unit_key,112        options=None113    ):114        data = {115            "units": [{"unit_key": unit_key, "type_id": type_id}],116            "options": options117        }118        return self.create_scheduled_action(pulp, action='/content/uninstall/', schedule=schedule, data=data)119    @property120    def certificate(self):121        return self.data['certificate']122    @certificate.setter123    def certificate(self, cert):124        self.data['certificate'] = cert125    @contextlib.contextmanager126    def tmp_certfile(self, closed=False):127        '''context manager creating a named temporary cert file'''128        import tempfile129        certfile = tempfile.NamedTemporaryFile(delete=False)130        certfile.write(self.certificate)131        certfile.seek(0)132        if closed:...

redshiftPausescheduler.py

Source:redshiftPausescheduler.py

...7        sys.exit(2)8    if str(action).lower() == 'create':9        try:10            client.describe_clusters(ClusterIdentifier=clusterName)11            client.create_scheduled_action(12                ScheduledActionName=scheduleName,13                TargetAction={14                    'ResumeCluster': {15                        'ClusterIdentifier': clusterName16                    }},17                Schedule=cronEntry,18                IamRole=iamRoleArn,19                Enable=True20            )21        except botoErr as error:22            print('my error ' + error.response['Error']['Message'])23    elif str(action).lower() == 'delete':24        DeleteSchedule(scheduleName)25    else:26        print('Invalid Action')27def ResizeSchedule(action, clusterName, scheduleName, iamRoleArn, cronEntry, nodeType, isClassicresize, resizeToNodes):28    client = boto3.client('redshift')29    if not clusterName or not scheduleName or not iamRoleArn or not cronEntry:30        sys.exit(2)31    if str(action).lower() == 'create':32        try:33            client.describe_clusters(ClusterIdentifier=clusterName)34            client.create_scheduled_action(35                ScheduledActionName=scheduleName,36                TargetAction={37                    'ResizeCluster': {38                        'ClusterIdentifier': clusterName,39                        'NodeType': nodeType,40                        'NumberOfNodes': resizeToNodes,41                        'Classic': isClassicresize42                    }},43                Schedule=cronEntry,44                IamRole=iamRoleArn,45                Enable=True46            )47        except botoErr as error:48            print('my error ' + error.response['Error']['Message'])...

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.