DM-35356: Change s3 to s3 copy implementation

python/lsst/resources/s3.py

@@ -42,15 +42,17 @@
     s3CheckFileExists,
 )
 
+try:
+    import boto3
+except ImportError:
+    boto3 = None
+
 try:
     from boto3.s3.transfer import TransferConfig  # type: ignore
 except ImportError:
     TransferConfig = None
 
 if TYPE_CHECKING:
-    with contextlib.suppress(ImportError):
-        import boto3
-
     from .utils import TransactionProtocol
 
 
@@ -333,13 +335,17 @@ def _upload_file(self, local_file: ResourcePath, progress: ProgressPercentage |
             raise
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
-    def _copy_from(self, src: ResourcePath) -> None:
+    def _copy_from(self, src: ResourcePath, progress: ProgressPercentage | None) -> None:
         copy_source = {
             "Bucket": src.netloc,
             "Key": src.relativeToPathRoot,
         }
+        assert boto3 is not None, "boto3 must be available to get here"
+        # Note that boto3.resource.meta.copy is cleverer than the low
+        # level copy_object.
+        s3 = boto3.resource("s3")
         try:
-            self.client.copy_object(CopySource=copy_source, Bucket=self.netloc, Key=self.relativeToPathRoot)
+            s3.meta.client.copy(copy_source, self.netloc, self.relativeToPathRoot, Callback=progress)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError("No such resource to transfer: {self}") from err
         except ClientError as err:
@@ -402,11 +408,14 @@ def transfer_from(
         timer_args = (src, self)
 
         if isinstance(src, type(self)):
-            # Looks like an S3 remote uri so we can use direct copy
-            # note that boto3.resource.meta.copy is cleverer than the low
-            # level copy_object
+            # Looks like an S3 remote uri so we can use direct copy.
+            progress = (
+                ProgressPercentage(src, msg="S3 to S3 copy:")
+                if log.isEnabledFor(ProgressPercentage.log_level)
+                else None
+            )
             with time_this(log, msg=timer_msg, args=timer_args):
-                self._copy_from(src)
+                self._copy_from(src, progress)
 
         else:
             # Use local file and upload it