Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use delete_objects method in localstack

Best Python code snippet using localstack_python

delete.py

Source:delete.py

1from . import Atomic2from . import util3import sys4from Atomic.backendutils import BackendUtils5ATOMIC_CONFIG = util.get_atomic_config()6storage = ATOMIC_CONFIG.get('default_storage', "docker")7class Delete(Atomic):8    def __init__(self):9        super(Delete, self).__init__()10        self.be = None11    def delete_image(self):12        """13        Mark given image(s) for deletion from registry14        :return: 0 if all images marked for deletion, otherwise 2 on any failure15        """16        if self.args.debug:17            util.write_out(str(self.args))18        if (len(self.args.delete_targets) > 0 and self.args.all) or (len(self.args.delete_targets) < 1 and not self.args.all):19            raise ValueError("You must select --all or provide a list of images to delete.")20        beu = BackendUtils()21        delete_objects = []22        # We need to decide on new returns for dbus because we now check image23        # validity prior to executing the delete.  If there is going to be a24        # failure, it will be here.25        #26        # The failure here is basically that it couldnt verify/find the image.27        if self.args.all:28            if self.args.storage:29                self.be = beu.get_backend_from_string(self.args.storage)30                delete_objects = self.be.get_images(get_all=True)31            else:32                delete_objects = beu.get_images(get_all=True)33        else:34            for image in self.args.delete_targets:35                _, img_obj = beu.get_backend_and_image_obj(image, str_preferred_backend=self.args.storage or storage, required=True if self.args.storage else False)36                delete_objects.append(img_obj)37        if self.args.remote:38            return self._delete_remote(self.args.delete_targets)39        if len(delete_objects) == 0:40            raise ValueError("No images to delete.")41        _image_names = []42        for del_obj in delete_objects:43            if del_obj.repotags:44                _image_names.append(len(del_obj.repotags[0]))45            else:46                _image_names.append(len(del_obj.id))47        max_img_name = max(_image_names) + 248        if not self.args.assumeyes:49            util.write_out("Do you wish to delete the following images?\n")50        else:51            util.write_out("The following images will be deleted.\n")52        two_col = "   {0:" + str(max_img_name) + "} {1}"53        util.write_out(two_col.format("IMAGE", "STORAGE"))54        for del_obj in delete_objects:55            image = None if not del_obj.repotags else del_obj.repotags[0]56            if image is None or "<none>" in image:57                image = del_obj.id[0:12]58            util.write_out(two_col.format(image, del_obj.backend.backend))59        if not self.args.assumeyes:60            confirm = util.input("\nConfirm (y/N) ")61            confirm = confirm.strip().lower()62            if not confirm in ['y', 'yes']:63                util.write_err("User aborted delete operation for {}".format(self.args.delete_targets or "all images"))64                sys.exit(2)65        # Perform the delete66        for del_obj in delete_objects:67            del_obj.backend.delete_image(del_obj.input_name, force=self.args.force)68        # We need to return something here for dbus69        return 070    def prune_images(self):71        """72        Remove dangling images from registry73        :return: 0 if all images deleted or no dangling images found74        """75        if self.args.debug:76            util.write_out(str(self.args))77        beu = BackendUtils()78        for backend in beu.available_backends:79            be = backend()80            be.prune()81        return 082    def _delete_remote(self, targets):83        results = 084        for target in targets:85            # _convert_to_skopeo requires registry v1 support while delete requires v2 support86            # args, img = self.syscontainers._convert_to_skopeo(target)87            args = []88            if "http:" in target:89                args.append("--insecure")90            for i in ["oci:", "http:", "https:"]:91                img = target.replace(i, "docker:")92            if not img.startswith("docker:"):93                img = "docker://" + img94            try:95                util.skopeo_delete(img, args)96                util.write_out("Image {} marked for deletion".format(img))97            except ValueError as e:98                util.write_err("Failed to mark Image {} for deletion: {}".format(img, e))99                results = 2...

delete_objects

Source:delete_objects

...39        'more_results': response['IsTruncated']40    }41    42    return keys43def delete_objects(client, bucket_name, prefix):44    some_results = get_keys_for_prefix(client, bucket_name, prefix)45    deleted_count = 046    while(some_results['more_results']):47        delete_objects = [{ 'Key': key } for key in some_results['objects']]48        49        if(len(delete_objects) >= 1):50            start_key = delete_objects[0]['Key'].split('/')[-1]51            end_key = delete_objects[-1]['Key'].split('/')[-1]52            print(f"Deleting {delete_objects[0]['Key'].split('/')[-1]} - {delete_objects[-1]['Key']}")53        54        response = client.delete_objects(55            Bucket=bucket_name,56            Delete={ 'Objects': delete_objects }57        )58        59        deleted_count += len(response['Deleted'])60        61        print(f"Deleted {deleted_count} objects")62        63        some_results = get_keys_for_prefix(client, bucket_name, prefix)64        65def main():66    parser = argparse.ArgumentParser()67    parser.add_argument("bucket", help="Bucket to delete from", type=str)68    parser.add_argument("prefix", help="Prefix to delete from", type=str)69    parser.add_argument("role_arn", help="Role ARN to run as", type=str)70    args = parser.parse_args()71    client = create_client('s3', args.role_arn)72        73    delete_objects(client, args.bucket, args.prefix)74    75if __name__ == "__main__":...

deleteSampleData.py

Source:deleteSampleData.py

...8    def handle(self, *args, **kwargs):9        from django.conf import settings10        if not settings.DEBUG:11            raise CommandError('You cannot run this command in production')12        self.delete_objects(models.AssetCategory)13        self.delete_objects(models.AssetStatus)14        self.delete_objects(models.Supplier)15        self.delete_objects(models.Connector)16        self.delete_objects(models.Asset)17        self.delete_objects(rigsmodels.VatRate)18        self.delete_objects(rigsmodels.Profile)19        self.delete_objects(rigsmodels.Person)20        self.delete_objects(rigsmodels.Organisation)21        self.delete_objects(rigsmodels.Venue)22        self.delete_objects(Group)23        self.delete_objects(rigsmodels.Event)24        self.delete_objects(rigsmodels.EventItem)25        self.delete_objects(rigsmodels.Invoice)26        self.delete_objects(rigsmodels.Payment)27        self.delete_objects(rigsmodels.RiskAssessment)28        self.delete_objects(rigsmodels.EventChecklist)29        self.delete_objects(tmodels.TrainingCategory)30        self.delete_objects(tmodels.TrainingItem)31        self.delete_objects(tmodels.TrainingLevel)32        self.delete_objects(tmodels.TrainingItemQualification)33        self.delete_objects(tmodels.TrainingLevelRequirement)34    def delete_objects(self, model):35        for obj in model.objects.all():...

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.