[v2] add parameter to prevent overwriting existing objects in S3 high level commands (#9750)

Add the `--no-overwrite` parameter to high-level AWS S3 operations `cp`, `mv`, and `sync`, which prevents overwriting objects that already exist at the destination.

This change makes use of the S3 feature `IfNoneMatch` which enforces conditional write operations on S3 buckets:

https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes-enforce.html

Updates have been applied to the vended `s3transfer` module as well as the AWS CLI to support the new parameter.

Author: kdaily

.changes/next-release/feature-s3-99392.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "``s3``",
+  "description": "Add the ``--no-overwrite`` parameter to high-level AWS S3 operations ``cp``, ``mv``, and ``sync``, which prevents overwriting objects that already exist at the destination. Implements `#2874 <https://github.com/aws/aws-cli/issues/2874>`__."
+}

awscli/customizations/s3/results.py

@@ -21,7 +21,10 @@
 
 from awscli.compat import ensure_text_type, queue
 from awscli.customizations.s3.subscribers import OnDoneFilteredSubscriber
-from awscli.customizations.s3.utils import WarningResult, human_readable_size
+from awscli.customizations.s3.utils import (
+    WarningResult,
+    human_readable_size,
+)
 from awscli.customizations.utils import uni_print
 
 LOGGER = logging.getLogger(__name__)
@@ -54,6 +57,7 @@ def _create_new_result_cls(name, extra_fields=None, base_cls=BaseResult):
 FailureResult = _create_new_result_cls('FailureResult', ['exception'])
 
 DryRunResult = _create_new_result_cls('DryRunResult')
+SkipFileResult = _create_new_result_cls('SkipFileResult')
 
 ErrorResult = namedtuple('ErrorResult', ['exception'])
 
@@ -123,6 +127,17 @@ def _on_failure(self, future, e):
             if isinstance(e, FatalError):
                 error_result_cls = ErrorResult
             self._result_queue.put(error_result_cls(exception=e))
+        elif self._is_precondition_failed(e):
+            LOGGER.debug(
+                f"warning: Skipping file {self._src} as it already exists on {self._dest}"
+            )
+            self._result_queue.put(
+                SkipFileResult(
+                    transfer_type=self._transfer_type,
+                    src=self._src,
+                    dest=self._dest,
+                )
+            )
         else:
             self._result_queue.put(
                 FailureResult(
@@ -133,6 +148,13 @@ def _on_failure(self, future, e):
                 )
             )
 
+    def _is_precondition_failed(self, exception):
+        return (
+            hasattr(exception, 'response')
+            and exception.response.get('Error', {}).get('Code')
+            == 'PreconditionFailed'
+        )
+
 
 class BaseResultHandler:
     """Base handler class to be called in the ResultProcessor"""
@@ -150,6 +172,7 @@ def __init__(self):
         self.files_transferred = 0
         self.files_failed = 0
         self.files_warned = 0
+        self.files_skipped = 0
         self.errors = 0
         self.expected_bytes_transferred = 0
         self.expected_files_transferred = 0
@@ -167,6 +190,7 @@ def __init__(self):
             SuccessResult: self._record_success_result,
             FailureResult: self._record_failure_result,
             WarningResult: self._record_warning_result,
+            SkipFileResult: self._record_skipped_file_result,
             ErrorResult: self._record_error_result,
             CtrlCResult: self._record_error_result,
             FinalTotalSubmissionsResult: self._record_final_expected_files,
@@ -282,6 +306,9 @@ def _record_failure_result(self, result, **kwargs):
         self.files_failed += 1
         self.files_transferred += 1
 
+    def _record_skipped_file_result(self, result, **kwargs):
+        self.files_skipped += 1
+
     def _record_warning_result(self, **kwargs):
         self.files_warned += 1
 
@@ -362,6 +389,7 @@ def __init__(
             SuccessResult: self._print_success,
             FailureResult: self._print_failure,
             WarningResult: self._print_warning,
+            SkipFileResult: self._print_skip,
             ErrorResult: self._print_error,
             CtrlCResult: self._print_ctrl_c,
             DryRunResult: self._print_dry_run,
@@ -380,6 +408,10 @@ def _print_noop(self, **kwargs):
         # If the result does not have a handler, then do nothing with it.
         pass
 
+    def _print_skip(self, **kwargs):
+        # Don't reset progress length since this result printer doesn't print a newline
+        self._redisplay_progress(reset_progress_length=False)
+
     def _print_dry_run(self, result, **kwargs):
         statement = self.DRY_RUN_FORMAT.format(
             transfer_type=result.transfer_type,
@@ -432,16 +464,19 @@ def _get_transfer_location(self, result):
             src=result.src, dest=result.dest
         )
 
-    def _redisplay_progress(self):
+    def _redisplay_progress(self, reset_progress_length=True):
         # Reset to zero because done statements are printed with new lines
         # meaning there are no carriage returns to take into account when
         # printing the next line.
-        self._progress_length = 0
+        if reset_progress_length:
+            self._progress_length = 0
         self._add_progress_if_needed()
 
     def _add_progress_if_needed(self):
         if self._has_remaining_progress():
             self._print_progress()
+        else:
+            self._clear_progress_if_no_more_expected_transfers(ending_char='\r')
 
     def _should_print_progress_now(self):
         """Check to see if should print progres based on frequency.
@@ -467,7 +502,7 @@ def _print_progress(self, **kwargs):
         remaining_files = self._get_expected_total(
             str(
                 self._result_recorder.expected_files_transferred
-                - self._result_recorder.files_transferred
+                - (self._result_recorder.files_transferred + self._result_recorder.files_skipped)
             )
         )
 
@@ -535,7 +570,7 @@ def _adjust_statement_padding(self, print_statement, ending_char='\n'):
     def _has_remaining_progress(self):
         if not self._result_recorder.expected_totals_are_final():
             return True
-        actual = self._result_recorder.files_transferred
+        actual = self._result_recorder.files_transferred + self._result_recorder.files_skipped
         expected = self._result_recorder.expected_files_transferred
         return actual != expected
 
@@ -545,9 +580,9 @@ def _print_to_out_file(self, statement):
     def _print_to_error_file(self, statement):
         uni_print(statement, self._error_file)
 
-    def _clear_progress_if_no_more_expected_transfers(self, **kwargs):
+    def _clear_progress_if_no_more_expected_transfers(self, ending_char='\n', **kwargs):
         if self._progress_length and not self._has_remaining_progress():
-            uni_print(self._adjust_statement_padding(''), self._out_file)
+            uni_print(self._adjust_statement_padding('', ending_char=ending_char), self._out_file)
 
 
 class NoProgressResultPrinter(ResultPrinter):

awscli/customizations/s3/s3handler.py

@@ -449,7 +449,36 @@ def _get_fileout(self, fileinfo):
         return fileinfo.dest
 
     def _get_warning_handlers(self):
-        return [self._warn_glacier, self._warn_parent_reference]
+        return [
+            self._warn_glacier,
+            self._warn_parent_reference,
+            self._warn_if_file_exists_with_no_overwrite,
+        ]
+
+    def _warn_if_file_exists_with_no_overwrite(self, fileinfo):
+        """
+        Warning handler to skip downloads when no-overwrite is set and local file exists.
+
+        This method prevents overwriting existing local files during S3 download operations
+        when the --no-overwrite flag is specified. It checks if the destination file already
+        exists on the local filesystem and skips the download if found.
+
+        :type fileinfo: FileInfo
+        :param fileinfo: The FileInfo object containing transfer details
+
+        :rtype: bool
+        :returns: True if the file should be skipped (exists and no-overwrite is set),
+                False if the download should proceed
+        """
+        if not self._cli_params.get('no_overwrite'):
+            return False
+        fileout = self._get_fileout(fileinfo)
+        if os.path.exists(fileout):
+            LOGGER.debug(
+                f"warning: skipping {fileinfo.src} -> {fileinfo.dest}, file exists at destination"
+            )
+            return True
+        return False
 
     def _format_src_dest(self, fileinfo):
         src = self._format_s3_path(fileinfo.src)
@@ -493,7 +522,9 @@ def _submit_transfer_request(self, fileinfo, extra_args, subscribers):
         )
 
     def _get_warning_handlers(self):
-        return [self._warn_glacier]
+        return [
+            self._warn_glacier,
+        ]
 
     def _format_src_dest(self, fileinfo):
         src = self._format_s3_path(fileinfo.src)

awscli/customizations/s3/subcommands.py

@@ -664,6 +664,15 @@
     ),
 }
 
+NO_OVERWRITE = {
+    'name': 'no-overwrite',
+    'action': 'store_true',
+    'help_text': (
+        "This flag prevents overwriting of files at the destination. With this flag, "
+        "only files not present at the destination will be transferred."
+    ),
+}
+
 
 CASE_CONFLICT = {
     'name': 'case-conflict',
@@ -1117,6 +1126,7 @@ class CpCommand(S3TransferCommand):
             EXPECTED_SIZE,
             RECURSIVE,
             CASE_CONFLICT,
+            NO_OVERWRITE,
         ]
     )
 
@@ -1142,6 +1152,7 @@ class MvCommand(S3TransferCommand):
             RECURSIVE,
             VALIDATE_SAME_S3_PATHS,
             CASE_CONFLICT,
+            NO_OVERWRITE,
         ]
     )
 
@@ -1187,7 +1198,7 @@ class SyncCommand(S3TransferCommand):
             }
         ]
         + TRANSFER_ARGS
-        + [METADATA, COPY_PROPS, METADATA_DIRECTIVE, CASE_CONFLICT]
+        + [METADATA, COPY_PROPS, METADATA_DIRECTIVE, CASE_CONFLICT, NO_OVERWRITE]
     )
 
 
@@ -1367,7 +1378,6 @@ def choose_sync_strategies(self):
                     sync_type = override_sync_strategy.sync_type
                     sync_type += '_sync_strategy'
                     sync_strategies[sync_type] = override_sync_strategy
-
         return sync_strategies
 
     def run(self):
@@ -1454,7 +1464,8 @@ def run(self):
             self._client, self._source_client, self.parameters
         )
 
-        s3_transfer_handler = S3TransferHandlerFactory(self.parameters)(
+        params = self._get_s3_handler_params()
+        s3_transfer_handler = S3TransferHandlerFactory(params)(
             self._transfer_manager, result_queue
         )
 
@@ -1575,6 +1586,14 @@ def _map_sse_c_params(self, request_parameters, paths_type):
                 },
             )
 
+    def _get_s3_handler_params(self):
+        params = self.parameters.copy()
+        
+        # Removing no-overwrite params from sync since file to be synced are
+        # already separated out using sync strategy
+        if self.cmd == 'sync':
+            params.pop('no_overwrite', None)
+        return params
     def _should_handle_case_conflicts(self):
         return (
             self.cmd in {'sync', 'cp', 'mv'}
@@ -1706,6 +1725,7 @@ def add_paths(self, paths):
         elif len(paths) == 1:
             self.parameters['dest'] = paths[0]
         self._validate_streaming_paths()
+        self._validate_no_overwrite_for_download_streaming()
         self._validate_path_args()
         self._validate_sse_c_args()
         self._validate_not_s3_express_bucket_for_sync()
@@ -1957,3 +1977,14 @@ def _validate_sse_c_copy_source_for_paths(self):
                     '--sse-c-copy-source is only supported for '
                     'copy operations.'
                 )
+
+    def _validate_no_overwrite_for_download_streaming(self):
+        if (
+            self.parameters['is_stream']
+            and self.parameters.get('no_overwrite')
+            and self.parameters['dest'] == '-'
+        ):
+            raise ParamValidationError(
+                "--no-overwrite parameter is not supported for "
+                "streaming downloads"
+            )

awscli/customizations/s3/syncstrategy/nooverwrite.py

@@ -0,0 +1,34 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+import logging
+
+from awscli.customizations.s3.subcommands import NO_OVERWRITE
+from awscli.customizations.s3.syncstrategy.base import BaseSync
+
+LOG = logging.getLogger(__name__)
+
+
+class NoOverwriteSync(BaseSync):
+    """Sync strategy that prevents overwriting of existing files at the destination.
+    This strategy is used only for files that exist at both source and destination
+    (file_at_src_and_dest_sync_strategy). It always returns False to prevent any
+    overwriting of existing files, regardless of size or modification time differences.
+    """
+
+    ARGUMENT = NO_OVERWRITE
+
+    def determine_should_sync(self, src_file, dest_file):
+        LOG.debug(
+            f"warning: skipping {src_file.src} -> {src_file.dest}, file exists at destination"
+        )
+        return False

awscli/customizations/s3/syncstrategy/register.py

@@ -14,6 +14,7 @@
 from awscli.customizations.s3.syncstrategy.exacttimestamps import (
     ExactTimestampsSync,
 )
+from awscli.customizations.s3.syncstrategy.nooverwrite import NoOverwriteSync
 from awscli.customizations.s3.syncstrategy.sizeonly import SizeOnlySync
 
 
@@ -48,4 +49,7 @@ def register_sync_strategies(command_table, session, **kwargs):
     # Register the delete sync strategy.
     register_sync_strategy(session, DeleteSync, 'file_not_at_src')
 
+    # Register the nooverwrite sync strategy
+    register_sync_strategy(session, NoOverwriteSync, 'file_at_src_and_dest')
+
     # Register additional sync strategies here...

awscli/customizations/s3/utils.py

@@ -489,6 +489,7 @@ def map_put_object_params(cls, request_params, cli_params):
         cls._set_sse_c_request_params(request_params, cli_params)
         cls._set_request_payer_param(request_params, cli_params)
         cls._set_checksum_algorithm_param(request_params, cli_params)
+        cls._set_no_overwrite_param(request_params, cli_params)
 
     @classmethod
     def map_get_object_params(cls, request_params, cli_params):
@@ -520,6 +521,7 @@ def map_copy_object_params(cls, request_params, cli_params):
         )
         cls._set_request_payer_param(request_params, cli_params)
         cls._set_checksum_algorithm_param(request_params, cli_params)
+        cls._set_no_overwrite_param(request_params, cli_params)
 
     @classmethod
     def map_head_object_params(cls, request_params, cli_params):
@@ -558,6 +560,11 @@ def map_delete_object_params(cls, request_params, cli_params):
     def map_list_objects_v2_params(cls, request_params, cli_params):
         cls._set_request_payer_param(request_params, cli_params)
 
+    @classmethod
+    def _set_no_overwrite_param(cls, request_params, cli_params):
+        if cli_params.get('no_overwrite'):
+            request_params['IfNoneMatch'] = "*"
+
     @classmethod
     def _set_request_payer_param(cls, request_params, cli_params):
         if cli_params.get('request_payer'):