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

How to use revoke_snapshot_access method in localstack

Best Python code snippet using localstack_python

client.py

Source:client.py

...172    def restore_table_from_cluster_snapshot(self, ClusterIdentifier: str, SnapshotIdentifier: str, SourceDatabaseName: str, SourceTableName: str, NewTableName: str, SourceSchemaName: str = None, TargetDatabaseName: str = None, TargetSchemaName: str = None) -> Dict:173        pass174    def revoke_cluster_security_group_ingress(self, ClusterSecurityGroupName: str, CIDRIP: str = None, EC2SecurityGroupName: str = None, EC2SecurityGroupOwnerId: str = None) -> Dict:175        pass176    def revoke_snapshot_access(self, SnapshotIdentifier: str, AccountWithRestoreAccess: str, SnapshotClusterIdentifier: str = None) -> Dict:177        pass178    def rotate_encryption_key(self, ClusterIdentifier: str) -> Dict:...

redshift_backup.py

Source:redshift_backup.py

...36			self.logger.exception(f"Failed t delete snapshot as the snapshot {backup_resource.backup_id} doesn't exist")37		if 'AccountsWithRestoreAccess' in snapshot:38			for shared_account in snapshot['AccountsWithRestoreAccess']:39				self.logger.info(f"revoking access to snapshot {backup_resource.backup_id} to account {shared_account['AccountId']} for the purpose of deletion")40				redshift_client.revoke_snapshot_access(41					SnapshotIdentifier=snapshot_id,42					SnapshotClusterIdentifier=cluster_id,43					AccountWithRestoreAccess=shared_account['AccountId']44				)45		try:46			redshift_client.delete_cluster_snapshot(47				SnapshotIdentifier=snapshot_id,48				SnapshotClusterIdentifier=cluster_id49			)50		except ClientError as ex:51			if 'other accounts still have access to it' in ex.response['Error']['Message']:52				self.logger.exception(f"Could not delete {backup_resource.backup_id} as "53				f"other accounts still have access to this snapshot")54				return...

redshift_delete_snapshot_action.py

Source:redshift_delete_snapshot_action.py

1######################################################################################################################2#  Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.                                           #3#                                                                                                                    #4#  Licensed under the Amazon Software License (the "License"). You may not use this file except in compliance        #5#  with the License. A copy of the License is located at                                                             #6#                                                                                                                    #7#      http://aws.amazon.com/asl/                                                                                    #8#                                                                                                                    #9#  or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES #10#  OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions    #11#  and limitations under the License.                                                                                #12######################################################################################################################13from datetime import datetime, timedelta14import dateutil.parser15import pytz16import services.redshift_service17from actions import *18from boto_retry import get_client_with_retries19INFO_REVOKE_ACCESS = "Revoking restore access for account {}"20INFO_DELETE_SNAPHOT = "Deleting snapshot {} for cluster {}"21GROUP_TITLE_DELETE_OPTIONS = "Snapshot delete options"22PARAM_DESC_RETENTION_COUNT = "Number of snapshots to keep for a RedShift cluster, use 0 to use retention days"23PARAM_DESC_RETENTION_DAYS = "Snapshot retention period in days, use 0 to use retention count"24PARAM_LABEL_RETENTION_COUNT = "Retention count"25PARAM_LABEL_RETENTION_DAYS = "Retention days"26INFO_ACCOUNT_SNAPSHOTS = "{} cluster snapshots for account {}"27INFO_KEEP_RETENTION_COUNT = "Retaining latest {} snapshots for each Redshift cluster"28INFO_REGION = "Processing snapshots in region {}"29INFO_RETENTION_DAYS = "Deleting snapshots older than {}"30INFO_SN_DELETE_RETENTION_COUNT = "Deleting snapshot {}, because count for its volume is {}"31INFO_SN_RETENTION_DAYS = "Deleting snapshot {} ({}) because it is older than retention period of {} days"32INFO_SNAPSHOT_DELETED = "Deleted snapshot {} for volume {}"33ERR_RETENTION_PARAM_BOTH = "Only one of {} or {} parameters can be specified"34ERR_RETENTION_PARAM_NONE = "{} or {} parameter must be specified"35PARAM_RETENTION_DAYS = "RetentionDays"36PARAM_RETENTION_COUNT = "RetentionCount"37class RedshiftDeleteSnapshotAction:38    properties = {39        ACTION_TITLE: "RedShift Delete Snapshot",40        ACTION_VERSION: "1.0",41        ACTION_DESCRIPION: "Deletes Redshift snapshots after retention period or count",42        ACTION_AUTHOR: "AWS",43        ACTION_ID: "2fb2442c-b847-4dab-b53e-e481e029cc30f",44        ACTION_SERVICE: "redshift",45        ACTION_RESOURCES: services.redshift_service.CLUSTER_SNAPSHOTS,46        ACTION_AGGREGATION: ACTION_AGGREGATION_ACCOUNT,47        ACTION_MEMORY: 128,48        ACTION_SELECT_EXPRESSION: "Snapshots[?Status=='available']|[?SnapshotType=='manual']."49                                  "{SnapshotIdentifier:SnapshotIdentifier, "50                                  "ClusterIdentifier:ClusterIdentifier,"51                                  "SnapshotCreateTime:SnapshotCreateTime,"52                                  "AccountsWithRestoreAccess:AccountsWithRestoreAccess[*].AccountId}",53        ACTION_KEEP_RESOURCE_TAGS: False,54        ACTION_ALLOW_TAGFILTER_WILDCARD: False,55        ACTION_PARAMETERS: {56            PARAM_RETENTION_DAYS: {57                PARAM_DESCRIPTION: PARAM_DESC_RETENTION_DAYS,58                PARAM_TYPE: type(0),59                PARAM_REQUIRED: False,60                PARAM_MIN_VALUE: 0,61                PARAM_LABEL: PARAM_LABEL_RETENTION_DAYS62            },63            PARAM_RETENTION_COUNT: {64                PARAM_DESCRIPTION: PARAM_DESC_RETENTION_COUNT,65                PARAM_TYPE: type(0),66                PARAM_REQUIRED: False,67                PARAM_MIN_VALUE: 0,68                PARAM_LABEL: PARAM_LABEL_RETENTION_COUNT69            }70        },71        ACTION_PARAMETER_GROUPS: [72            {73                ACTION_PARAMETER_GROUP_TITLE: GROUP_TITLE_DELETE_OPTIONS,74                ACTION_PARAMETER_GROUP_LIST: [75                    PARAM_RETENTION_DAYS,76                    PARAM_RETENTION_COUNT77                ],78            }],79        ACTION_PERMISSIONS: ["redshift:DescribeClusterSnapshots",80                             "redshift:DeleteClusterSnapshot",81                             "redshift:RevokeSnapshotAccess"]82    }83    @staticmethod84    def action_validate_parameters(parameters):85        retention_days = parameters.get(PARAM_RETENTION_DAYS)86        retention_count = parameters.get(PARAM_RETENTION_COUNT)87        if not retention_count and not retention_days:88            raise ValueError(ERR_RETENTION_PARAM_NONE.format(PARAM_RETENTION_COUNT, PARAM_RETENTION_DAYS))89        if retention_days and retention_count:90            raise ValueError(ERR_RETENTION_PARAM_BOTH.format(PARAM_RETENTION_COUNT, PARAM_RETENTION_DAYS))91        return parameters92    def __init__(self, arguments):93        self._arguments = arguments94        self.context = self._arguments[ACTION_PARAM_CONTEXT]95        self.logger = self._arguments[ACTION_PARAM_LOGGER]96        self.task = self._arguments[ACTION_PARAM_TASK]97        self.snapshots = sorted(self._arguments[ACTION_PARAM_RESOURCES])98        self.retention_days = self._arguments.get(PARAM_RETENTION_DAYS)99        self.retention_count = self._arguments.get(PARAM_RETENTION_COUNT)100        self.dryrun = self._arguments.get(ACTION_PARAM_DRYRUN, False)101        self.session = self._arguments[ACTION_PARAM_SESSION]102        self.account = self.snapshots[0]["AwsAccount"]103        self.result = {104            "account": self.account,105            "task": self.task106        }107    def execute(self, _):108        def snapshots_to_delete():109            def by_retention_days():110                delete_before_dt = datetime.utcnow().replace(tzinfo=pytz.timezone("UTC")) - timedelta(days=int(self.retention_days))111                self.logger.info(INFO_RETENTION_DAYS, delete_before_dt)112                for sn in sorted(self.snapshots, key=lambda s: s["Region"]):113                    snapshot_dt = dateutil.parser.parse(sn["SnapshotCreateTime"])114                    if snapshot_dt < delete_before_dt:115                        self.logger.info(INFO_SN_RETENTION_DAYS, sn["SnapshotIdentifier"], sn["SnapshotCreateTime"],116                                         self.retention_days)117                        yield sn118            def by_retention_count():119                self.logger.info(INFO_KEEP_RETENTION_COUNT, self.retention_count)120                sorted_snapshots = sorted(self.snapshots,121                                          key=lambda s: (s["ClusterIdentifier"], dateutil.parser.parse(s["SnapshotCreateTime"])),122                                          reverse=True)123                volume = None124                count_for_volume = 0125                for sn in sorted_snapshots:126                    if sn["ClusterIdentifier"] != volume:127                        volume = sn["ClusterIdentifier"]128                        count_for_volume = 0129                    count_for_volume += 1130                    if count_for_volume > self.retention_count:131                        self.logger.info(INFO_SN_DELETE_RETENTION_COUNT, sn["SnapshotIdentifier"], count_for_volume)132                        yield sn133            return by_retention_days() if self.retention_days else by_retention_count()134        self.logger.info("{}, version {}", self.properties[ACTION_TITLE], self.properties[ACTION_VERSION])135        region = None136        redshift = None137        deleted_count = 0138        self.logger.info(INFO_ACCOUNT_SNAPSHOTS, len(self.snapshots), self.account)139        self.logger.debug("Cluster Snapshots : {}", self.snapshots)140        for snapshot in snapshots_to_delete():141            if snapshot["Region"] != region:142                region = snapshot["Region"]143                self.logger.info(INFO_REGION, region)144                redshift = get_client_with_retries("redshift", ["delete_cluster_snapshot", "revoke_snapshot_access"], region=region,145                                                   context=self.context, session=self.session)146                if "deleted" not in self.result:147                    self.result["deleted"] = {}148                self.result["deleted"][region] = []149            try:150                snapshot_id = snapshot["SnapshotIdentifier"]151                cluster_id = snapshot["ClusterIdentifier"]152                granted_accounts = snapshot.get("AccountsWithRestoreAccess", [])153                if granted_accounts is None:154                    granted_accounts = []155                self.logger.info(INFO_DELETE_SNAPHOT, snapshot_id, cluster_id)156                for account in granted_accounts:157                    self.logger.info(INFO_REVOKE_ACCESS, account)158                    redshift.revoke_snapshot_access_with_retries(SnapshotIdentifier=snapshot_id,159                                                                 SnapshotClusterIdentifier=cluster_id,160                                                                 AccountWithRestoreAccess=account)161                redshift.delete_cluster_snapshot_with_retries(SnapshotIdentifier=snapshot_id,162                                                              SnapshotClusterIdentifier=cluster_id)163                deleted_count += 1164                self.logger.info(INFO_SNAPSHOT_DELETED, snapshot_id, cluster_id)165                self.result["deleted"][region].append(snapshot_id)166            except Exception as ex:167                if self.dryrun:168                    self.logger.debug(str(ex))169                    self.result["delete_cluster_snapshot"] = str(ex)170                    return self.result171                else:172                    raise ex173        self.result.update({174            "snapshots": len(self.snapshots),175            "total-deleted": deleted_count176        })...

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.