Merge branch 'release-0.19.0'

* release-0.19.0:
  Bumping version to 0.19.0
  Merge customizations for S3
  Bump aws-actions/configure-aws-credentials in the github-actions group (#395)
  Bump https://github.com/astral-sh/ruff-pre-commit (#394)
  Bump github/codeql-action in the github-actions group (#393)

Author: aws-sdk-python-automation

.changes/0.19.0.json

@@ -0,0 +1,12 @@
+[
+  {
+    "category": "``s3``",
+    "description": "Update multi-part copy logic to match single-part behavior for TaggingDirective and AnnotationDirective",
+    "type": "feature"
+  },
+  {
+    "category": "``s3``",
+    "description": "Warn when Metadata or Tagging is supplied to a copy without the corresponding directive set to REPLACE, in which case the supplied value is silently ignored. This matches the CopyObject behavior.",
+    "type": "enhancement"
+  }
+]

.github/workflows/codeql.yml

@@ -25,13 +25,13 @@ jobs:
         persist-credentials: false
 
     - name: "Run CodeQL init"
-      uses: "github/codeql-action/init@95e58e9a2cdfd71adc6e0353d5c52f41a045d225"
+      uses: "github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa"
       with:
         config-file: "./.github/codeql.yml"
         languages: "python"
 
     - name: "Run CodeQL autobuild"
-      uses: "github/codeql-action/autobuild@95e58e9a2cdfd71adc6e0353d5c52f41a045d225"
+      uses: "github/codeql-action/autobuild@7211b7c8077ea37d8641b6271f6a365a22a5fbfa"
 
     - name: "Run CodeQL analyze"
-      uses: "github/codeql-action/analyze@95e58e9a2cdfd71adc6e0353d5c52f41a045d225"
+      uses: "github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa"

.github/workflows/pull-request-build.yml

@@ -26,7 +26,7 @@ jobs:
       contents: read
     steps:
       - name: Configure AWS Credentials
-        uses: aws-actions/configure-aws-credentials@ec61189d14ec14c8efccab744f656cffd0e33f37
+        uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b
         with:
           role-to-assume: ${{ env.IAM_ROLE_ARN }}
           role-session-name: PullRequestBuildGitHubAction

.pre-commit-config.yaml

@@ -7,7 +7,7 @@ repos:
       - id: end-of-file-fixer
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 6fec9b7edb08fd9989088709d864a7826dc74e80  # frozen: v0.15.12
+    rev: 22f0422809455ec89ffdcf5a00170ba816e42ddb  # frozen: v0.15.16
     hooks:
       - id: ruff-check
         args: [ --fix ]

CHANGELOG.rst

@@ -2,6 +2,13 @@
 CHANGELOG
 =========
 
+0.19.0
+======
+
+* feature:``s3``: Update multi-part copy logic to match single-part behavior for TaggingDirective and AnnotationDirective
+* enhancement:``s3``: Warn when Metadata or Tagging is supplied to a copy without the corresponding directive set to REPLACE, in which case the supplied value is silently ignored. This matches the CopyObject behavior.
+
+
 0.18.0
 ======
 

s3transfer/__init__.py

@@ -145,7 +145,7 @@ def __call__(self, bytes_amount):
 from s3transfer.exceptions import RetriesExceededError, S3UploadFailedError
 
 __author__ = 'Amazon Web Services'
-__version__ = '0.18.0'
+__version__ = '0.19.0'
 
 
 logger = logging.getLogger(__name__)

s3transfer/copies.py

@@ -12,6 +12,7 @@
 # language governing permissions and limitations under the License.
 import copy
 import math
+from urllib.parse import parse_qsl
 
 from botocore.exceptions import ClientError
 
@@ -70,6 +71,8 @@ class CopySubmissionTask(SubmissionTask):
         'CopySourceSSECustomerKeyMD5',
         'MetadataDirective',
         'TaggingDirective',
+        'AnnotationDirective',
+        'Tagging',
     ]
 
     # Metadata fields to preserve for multipart copies.
@@ -91,6 +94,20 @@ class CopySubmissionTask(SubmissionTask):
         'ExpectedBucketOwner',
     ]
 
+    GET_OBJECT_TAGGING_ARGS = ['RequestPayer', 'ExpectedBucketOwner']
+    PUT_OBJECT_TAGGING_ARGS = [
+        'RequestPayer',
+        'ExpectedBucketOwner',
+        'ChecksumAlgorithm',
+    ]
+    LIST_OBJECT_ANNOTATIONS_ARGS = ['RequestPayer', 'ExpectedBucketOwner']
+    GET_OBJECT_ANNOTATION_ARGS = ['RequestPayer', 'ExpectedBucketOwner']
+    PUT_OBJECT_ANNOTATION_ARGS = [
+        'RequestPayer',
+        'ExpectedBucketOwner',
+        'ChecksumAlgorithm',
+    ]
+
     def _submit(
         self, client, config, osutil, request_executor, transfer_future
     ):
@@ -113,6 +130,8 @@ def _submit(
             transfer request that tasks are being submitted for
         """
         preserved_metadata = {}
+        source_version_id = None
+        call_args = transfer_future.meta.call_args
         if (
             transfer_future.meta.size is None
             or transfer_future.meta.etag is None
@@ -122,7 +141,6 @@ def _submit(
             # the TransferManager. If the object is outside of the region
             # of the client, they may have to provide the file size themselves
             # with a completely new client.
-            call_args = transfer_future.meta.call_args
             head_object_request = (
                 self._get_head_object_request_from_copy_source(
                     call_args.copy_source
@@ -148,6 +166,9 @@ def _submit(
             # during a multipart copy.
             transfer_future.meta.provide_object_etag(response.get('ETag'))
             preserved_metadata = self._extract_preserved_metadata(response)
+            # Pin the source version so all subsequent reads (tags, annotations)
+            # are consistent with the object from the head call
+            source_version_id = response.get('VersionId')
 
         # If it is greater than threshold do a multipart copy, otherwise
         # do a regular copy object.
@@ -163,6 +184,7 @@ def _submit(
                 request_executor,
                 transfer_future,
                 preserved_metadata,
+                source_version_id=source_version_id,
             )
 
     def _submit_copy_request(
@@ -199,9 +221,10 @@ def _submit_multipart_request(
         request_executor,
         transfer_future,
         preserved_metadata=None,
+        source_version_id=None,
     ):
         call_args = transfer_future.meta.call_args
-        merged_extra_args = self._merge_preserved_metadata(
+        merged_extra_args = self._apply_preserved_metadata(
             call_args.extra_args, preserved_metadata or {}
         )
 
@@ -293,16 +316,19 @@ def _submit_multipart_request(
         complete_multipart_extra_args = self._extra_complete_multipart_args(
             call_args.extra_args
         )
+
         # Submit the request to complete the multipart upload.
         self._transfer_coordinator.submit(
             request_executor,
-            CompleteMultipartUploadTask(
+            CopyCompleteMultipartUploadTask(
                 transfer_coordinator=self._transfer_coordinator,
                 main_kwargs={
                     'client': client,
                     'bucket': call_args.bucket,
                     'key': call_args.key,
                     'extra_args': complete_multipart_extra_args,
+                    'call_args': call_args,
+                    'source_version_id': source_version_id,
                 },
                 pending_main_kwargs={
                     'upload_id': create_multipart_future,
@@ -319,15 +345,18 @@ def _extract_preserved_metadata(self, head_object_response):
                 preserved[field] = head_object_response[field]
         return preserved
 
-    def _merge_preserved_metadata(self, extra_args, preserved_metadata):
-        if not preserved_metadata:
-            return extra_args
+    def _apply_preserved_metadata(self, extra_args, preserved_metadata):
+        # MPU has no native MetadataDirective, handle metadata manually.  REPLACE
+        # means we copy whatever the user provided, anything else means we drop
+        # what the user supplied
         if extra_args.get('MetadataDirective') == 'REPLACE':
             return extra_args
-        merged = dict(extra_args)
-        for field, value in preserved_metadata.items():
-            merged[field] = value
-        return merged
+        result = {
+            k: v for k, v in extra_args.items()
+            if k not in self.PRESERVED_METADATA_FIELDS
+        }
+        result.update(preserved_metadata)
+        return result
 
     def _get_head_object_request_from_copy_source(self, copy_source):
         if isinstance(copy_source, dict):
@@ -357,6 +386,148 @@ def _get_transfer_size(
         return part_size
 
 
+class CopyCompleteMultipartUploadTask(CompleteMultipartUploadTask):
+    """CompleteMultipartUpload variant that also applies tags and annotations.
+
+    After the destination object is finalized, copies/applies tags and
+    annotations inline. Errors during apply propagate as task failures.
+    """
+
+    def _main(
+        self,
+        client,
+        bucket,
+        key,
+        upload_id,
+        parts,
+        extra_args,
+        call_args,
+        source_version_id,
+    ):
+        response = client.complete_multipart_upload(
+            Bucket=bucket,
+            Key=key,
+            UploadId=upload_id,
+            MultipartUpload={'Parts': parts},
+            **extra_args,
+        )
+        dest_etag = response.get('ETag')
+        dest_version_id = response.get('VersionId')
+        self._apply_tags(client, call_args, source_version_id, dest_version_id)
+        self._apply_annotations(
+            client, call_args, source_version_id, dest_version_id, dest_etag
+        )
+
+    def _apply_tags(self, client, call_args, source_version_id, dest_version_id):
+        extra_args = call_args.extra_args
+        directive = extra_args.get('TaggingDirective')
+        if directive not in ('COPY', 'REPLACE'):
+            return
+        if directive == 'COPY':
+            src_kwargs = {
+                'Bucket': call_args.copy_source['Bucket'],
+                'Key': call_args.copy_source['Key'],
+                **get_filtered_dict(
+                    extra_args, CopySubmissionTask.GET_OBJECT_TAGGING_ARGS
+                ),
+            }
+            if source_version_id:
+                src_kwargs['VersionId'] = source_version_id
+            tag_set = call_args.source_client.get_object_tagging(
+                **src_kwargs
+            ).get('TagSet', [])
+        else:  # REPLACE
+            tag_set = [
+                {'Key': k, 'Value': v}
+                for k, v in parse_qsl(
+                    extra_args.get('Tagging', ''),
+                    keep_blank_values=True,
+                )
+            ]
+        if not tag_set:
+            return
+        put_kwargs = {
+            'Bucket': call_args.bucket,
+            'Key': call_args.key,
+            'Tagging': {'TagSet': tag_set},
+            **get_filtered_dict(
+                extra_args, CopySubmissionTask.PUT_OBJECT_TAGGING_ARGS
+            ),
+        }
+        if dest_version_id:
+            put_kwargs['VersionId'] = dest_version_id
+        client.put_object_tagging(**put_kwargs)
+
+    def _apply_annotations(
+        self,
+        client,
+        call_args,
+        source_version_id,
+        dest_version_id,
+        dest_etag,
+    ):
+        # We copy annotations only if COPY is explicitly set by the user.
+        extra_args = call_args.extra_args
+        if extra_args.get('AnnotationDirective') != 'COPY':
+            return
+        src_base = {
+            'Bucket': call_args.copy_source['Bucket'],
+            'Key': call_args.copy_source['Key'],
+        }
+        if source_version_id:
+            src_base['VersionId'] = source_version_id
+        list_kwargs = {
+            **src_base,
+            **get_filtered_dict(
+                extra_args, CopySubmissionTask.LIST_OBJECT_ANNOTATIONS_ARGS
+            ),
+        }
+        get_kwargs_base = {
+            **src_base,
+            **get_filtered_dict(
+                extra_args, CopySubmissionTask.GET_OBJECT_ANNOTATION_ARGS
+            ),
+        }
+        put_passthrough = get_filtered_dict(
+            extra_args, CopySubmissionTask.PUT_OBJECT_ANNOTATION_ARGS
+        )
+        list_response = call_args.source_client.list_object_annotations(
+            **list_kwargs
+        )
+        succeeded = []
+        failed = {}
+        for annotation in list_response.get('Annotations', []):
+            name = annotation['AnnotationName']
+            payload_response = call_args.source_client.get_object_annotation(
+                **get_kwargs_base,
+                AnnotationName=name,
+            )
+            put_kwargs = {
+                'Bucket': call_args.bucket,
+                'Key': call_args.key,
+                'AnnotationName': name,
+                'AnnotationPayload': payload_response['AnnotationPayload'].read(),
+                **put_passthrough,
+            }
+            if dest_version_id:
+                put_kwargs['VersionId'] = dest_version_id
+            if dest_etag:
+                put_kwargs['ObjectIfMatch'] = dest_etag
+            try:
+                client.put_object_annotation(**put_kwargs)
+                succeeded.append(name)
+            except Exception as e:
+                failed[name] = e
+        if failed:
+            raise S3CopyFailedError(
+                f'Failed to copy annotations to '
+                f's3://{call_args.bucket}/{call_args.key}. '
+                f'Succeeded: {succeeded}. '
+                f'Failed: {list(failed.keys())}. '
+                f'Errors: {failed}'
+            )
+
+
 class CopyObjectTask(Task):
     """Task to do a nonmultipart copy"""
 

s3transfer/manager.py

@@ -220,6 +220,7 @@ class TransferManager:
         'CopySourceSSECustomerKeyMD5',
         'MetadataDirective',
         'TaggingDirective',
+        'AnnotationDirective',
     ]
 
     ALLOWED_DELETE_ARGS = [
@@ -456,6 +457,25 @@ def copy(
             subscribers = []
         if source_client is None:
             source_client = self._client
+        # Warn when Metadata/Tagging are supplied without a directive. To match
+        # the low-level CopyObject behavior, the supplied values are silently
+        # ignored unless the corresponding directive is set to 'REPLACE'. The
+        # warning surfaces this so callers don't get blindsided when their
+        # input has no effect.
+        if extra_args.get('Metadata') and extra_args.get('MetadataDirective') is None:
+            logger.warning(
+                "Metadata was supplied without a metadata directive. The "
+                "supplied metadata will be ignored and source metadata will "
+                "be preserved. Set the metadata directive to 'REPLACE' to "
+                "apply the supplied metadata."
+            )
+        if extra_args.get('Tagging') and extra_args.get('TaggingDirective') is None:
+            logger.warning(
+                "Tagging was supplied without a tagging directive. The "
+                "supplied tagging will be ignored and source tags will be "
+                "preserved. Set the tagging directive to 'REPLACE' to apply "
+                "the supplied tagging."
+            )
         self._validate_all_known_args(extra_args, self.ALLOWED_COPY_ARGS)
         if isinstance(copy_source, dict):
             self._validate_if_bucket_supported(copy_source.get('Bucket'))