Handling versioning of object

Updated comments

Author: Rakshil Modi

awscli/customizations/s3/filegenerator.py

@@ -19,13 +19,16 @@
 from dateutil.tz import tzlocal
 
 from awscli.compat import queue
+from awscli.customizations.s3.fileinfo import VersionedFileInfo
 from awscli.customizations.s3.utils import (
     EPOCH_TIME,
     BucketLister,
+    BucketVersionLister,
     create_warning,
     find_bucket_key,
     find_dest_path_comp_key,
     get_file_stat,
+    split_s3_bucket_key,
 )
 
 _open = open
@@ -406,3 +409,114 @@ def _list_single_object(self, s3_path):
         response['LastModified'] = last_update.astimezone(tzlocal())
         response['ETag'] = response.pop('ETag', None)
         return s3_path, response
+
+
+class VersionedFileGenerator:
+    """
+    This class generates VersionedFileInfo objects for all versions of objects in a bucket.
+    It uses the BucketVersionLister to list all versions and creates appropriate
+    VersionedFileInfo objects for each version.
+    """
+
+    def __init__(
+        self,
+        client,
+        operation_name,
+        follow_symlinks=True,
+        page_size=None,
+        result_queue=None,
+        request_parameters=None,
+    ):
+        """
+        Initialize a new VersionedFileGenerator.
+
+        :param client: The S3 client to use.
+        :param operation_name: The name of the operation to perform.
+        :param follow_symlinks: Whether to follow symlinks.
+        :param page_size: The number of items to include in each API response.
+        :param result_queue: Queue for results and warnings.
+        :param request_parameters: Additional parameters for the request.
+        """
+        self._client = client
+        self.operation_name = operation_name
+        self.follow_symlinks = follow_symlinks
+        self.page_size = page_size
+        self.result_queue = result_queue
+        if not result_queue:
+            self.result_queue = queue.Queue()
+        self.request_parameters = {}
+        if request_parameters is not None:
+            self.request_parameters = request_parameters
+        self._version_lister = BucketVersionLister(client)
+
+    def call(self, files):
+        """
+        Generate VersionedFileInfo objects for all versions of objects.
+
+        :param files: Dictionary containing source and destination information.
+        :yields: VersionedFileInfo objects for each version of each object.
+        """
+        source = files['src']['path']
+        src_type = files['src']['type']
+        dest_type = files['dest']['type']
+
+        # Use the list_object_versions method to get all versions
+        file_iterator = self.list_object_versions(source, files['dir_op'])
+
+        for src_path, content, version_id in file_iterator:
+            dest_path, compare_key = find_dest_path_comp_key(files, src_path)
+
+            # Check if this is a delete marker
+            is_delete_marker = content.get('DeleteMarker', False)
+
+            # Create a VersionedFileInfo for this object version
+            yield VersionedFileInfo(
+                src=src_path,
+                dest=dest_path,
+                compare_key=compare_key,
+                size=content.get('Size', 0),
+                last_update=content.get('LastModified'),
+                src_type=src_type,
+                dest_type=dest_type,
+                operation_name=self.operation_name,
+                # client=self._client,
+                associated_response_data=content,
+                version_id=version_id,
+                is_delete_marker=is_delete_marker,
+            )
+
+    def list_object_versions(self, s3_path, dir_op):
+        """
+        This function yields the appropriate object versions or all object versions
+        under a common prefix depending if the operation is on objects under a
+        common prefix. It yields the file's source path, content, and version ID.
+
+        :param s3_path: The S3 path to list versions for.
+        :param dir_op: Whether this is a directory operation.
+        :yields: Tuples of (source_path, content, version_id)
+        """
+        bucket, key = split_s3_bucket_key(s3_path)
+
+        # Short circuit path: if we are not recursing into the s3
+        # bucket and a specific path was given, we can just yield
+        # that path and not have to call any operation in s3.
+        # However, for versioned objects, we still need to list all versions
+        # even for a specific object, so we don't have a short circuit path here.
+
+        # List all versions of objects
+        for (
+            src_path,
+            content,
+            version_id,
+        ) in self._version_lister.list_object_versions(
+            bucket=bucket,
+            prefix=key,
+            page_size=self.page_size,
+            extra_args=self.request_parameters.get('ListObjectVersions', {}),
+        ):
+            # If this is not a directory operation and the path doesn't match exactly,
+            # skip it (similar to the behavior in FileGenerator.list_objects)
+            if not dir_op and s3_path != src_path:
+                continue
+
+            yield src_path, content, version_id

awscli/customizations/s3/fileinfo.py

@@ -109,3 +109,40 @@ def _is_restored(self, response_data):
         # restored back to S3.
         # 'Restore' looks like: 'ongoing-request="false", expiry-date="..."'
         return 'ongoing-request="false"' in response_data.get('Restore', '')
+
+
+class VersionedFileInfo(FileInfo):
+    def __init__(self, version_id=None, is_delete_marker=False, **kwargs):
+        """
+        This class extends FileInfo to include version information for S3 objects.
+        It is specifically designed for operations that need to work with versioned
+        S3 objects, such as deleting all versions of an object.
+
+        :param version_id: The version ID of the S3 object.
+        :type version_id: string
+        :param is_delete_marker: Whether this version is a delete marker.
+        :type is_delete_marker: boolean
+        """
+        super().__init__(**kwargs)
+        self.version_id = version_id
+        self.is_delete_marker = is_delete_marker
+
+    def is_glacier_compatible(self):
+        """
+        Determines glacier compatibility for versioned S3 objects, with special handling for delete operations.
+
+        This method overrides the parent FileInfo.is_glacier_compatible() to provide enhanced
+        compatibility checking for S3 objects stored in glacier storage classes
+        when versioning is enabled on the bucket.
+
+        This method override allows delete operations to proceed on versioned glacier objects.
+        Since delete operations on glacier objects succeed regardless of storage class
+
+        :rtype: bool
+        :returns: True if the operation can proceed on glacier objects (specifically for delete
+                operations on versioned objects), False if the operation would fail due to
+                glacier storage class restrictions
+        """
+        if self.operation_name == 'delete':
+            return True
+        return super().is_glacier_compatible()