KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use restore_from_cluster_snapshot method in localstack

Best Python code snippet using localstack_python

backupredshift.py

Source:backupredshift.py

...25     if mybackupname == mysnapshot['SnapshotIdentifier'] :26          snapshot_exist=127          print 'Snapshot ' + mybackupname + ' exist.'28#4. start the redshift cluster from a snapshot29# restore_from_cluster_snapshot(cluster_identifier, snapshot_identifier, snapshot_cluster_identifier=None,30#  port=None, availability_zone=None, allow_version_upgrade=None, cluster_subnet_group_name=None, publicly_accessible=None,31#  owner_account=None, hsm_client_certificate_identifier=None, hsm_configuration_identifier=None, elastic_ip=None,32#  cluster_parameter_group_name=None, cluster_security_groups=None, vpc_security_group_ids=None,33#  preferred_maintenance_window=None, automated_snapshot_retention_period=None)34def start_cluster():35     global myidentifier, mybackupname, myelastic_ip, cluster_exist, snapshot_exist, conn, mydict36     try:37          # if there is no snapshot, we cannot start this redshift cluster, just exit the program.38          if snapshot_exist == 0 :39               print 'Warning! Snapshot ' + mybackupname + ' does NOT exist! Exit...'40               import sys41               sys.exit()42          myclusterresponse=mydict['DescribeClustersResponse']['DescribeClustersResult']['Clusters']43          for mycluster in myclusterresponse:44               #print mycluster['ClusterIdentifier']45               if myidentifier == mycluster['ClusterIdentifier'] :46                    cluster_exist = 147          if cluster_exist == 1 :48               print 'The Cluster ' + myidentifier + ' is already up!'49          else:50               # comment it out for testing51               conn.restore_from_cluster_snapshot(myidentifier, mybackupname, availability_zone='us-east-1a', elastic_ip=myelastic_ip)52               print 'start Cluster ' + myidentifier53     except IndexError:54          # comment it out for testing55          conn.restore_from_cluster_snapshot(myidentifier, mybackupname, availability_zone='us-east-1a', elastic_ip=myelastic_ip)56          print 'start Cluster ' + myidentifier57#5. stop the redshift cluster. ie. delete the redshift cluster after creaing a final snapshot58# delete_cluster(cluster_identifier, skip_final_cluster_snapshot=None, final_cluster_snapshot_identifier=None)59def stop_cluster():60     global myidentifier, mybackupname, myelastic_ip, cluster_exist, snapshot_exist, conn, mydict61     try:62          # Note the case that the cluster is already stopped!63          # The lenth is zero if no redshift cluster is running.64          if len(mydict['DescribeClustersResponse']['DescribeClustersResult']['Clusters']) :65               print 'you can stop.'66               #5.1 delete the snapshot testbackup67               # What if the cluster is already in the stop state?68               # comment it out for testing69               # delete_cluster_snapshot(snapshot_identifier, snapshot_cluster_identifier=None)...

redshift.py

Source:redshift.py

...75        snapshots = response['Snapshots']76        snapshots = [snapshot for snapshot in snapshots if snapshot["Status"]]77        snapshots.sort(key=lambda x: x['SnapshotCreateTime'], reverse=True)78        return snapshots79    def restore_from_cluster_snapshot(self, cluster_identifier: str, snapshot_identifier: str) -> str:80        """81        Restores a cluster from its snapshot82        :param cluster_identifier: unique identifier of a cluster83        :type cluster_identifier: str84        :param snapshot_identifier: unique identifier for a snapshot of a cluster85        :type snapshot_identifier: str86        """87        response = self.get_conn().restore_from_cluster_snapshot(88            ClusterIdentifier=cluster_identifier, SnapshotIdentifier=snapshot_identifier89        )90        return response['Cluster'] if response['Cluster'] else None91    def create_cluster_snapshot(self, snapshot_identifier: str, cluster_identifier: str) -> str:92        """93        Creates a snapshot of a cluster94        :param snapshot_identifier: unique identifier for a snapshot of a cluster95        :type snapshot_identifier: str96        :param cluster_identifier: unique identifier of a cluster97        :type cluster_identifier: str98        """99        response = self.get_conn().create_cluster_snapshot(100            SnapshotIdentifier=snapshot_identifier,101            ClusterIdentifier=cluster_identifier,...

redshift_hook.py

Source:redshift_hook.py

...71        snapshots = response['Snapshots']72        snapshots = filter(lambda x: x['Status'], snapshots)73        snapshots.sort(key=lambda x: x['SnapshotCreateTime'], reverse=True)74        return snapshots75    def restore_from_cluster_snapshot(self, cluster_identifier, snapshot_identifier):76        """77        Restores a cluster from its snapshot78        :param cluster_identifier: unique identifier of a cluster79        :type cluster_identifier: str80        :param snapshot_identifier: unique identifier for a snapshot of a cluster81        :type snapshot_identifier: str82        """83        response = self.get_conn().restore_from_cluster_snapshot(84            ClusterIdentifier=cluster_identifier,85            SnapshotIdentifier=snapshot_identifier86        )87        return response['Cluster'] if response['Cluster'] else None88    def create_cluster_snapshot(self, snapshot_identifier, cluster_identifier):89        """90        Creates a snapshot of a cluster91        :param snapshot_identifier: unique identifier for a snapshot of a cluster92        :type snapshot_identifier: str93        :param cluster_identifier: unique identifier of a cluster94        :type cluster_identifier: str95        """96        response = self.get_conn().create_cluster_snapshot(97            SnapshotIdentifier=snapshot_identifier,...

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.