Add cross-provider ranged, parallel downloads

CHANGELOG.rst

@@ -1,3 +1,20 @@
+4.3.0 - unreleased
+------------------
+
+## What's new
+* **Cross-provider ranged, parallel downloads.** The new
+  ``BucketObject.download_to_file`` fetches objects larger than the configured
+  threshold as ranged reads of ``part_size`` bytes, up to ``max_concurrency``
+  parts in parallel, so large downloads are no longer bound to a single
+  connection (and the whole object is never held in memory). AWS delegates to
+  boto3's TransferManager and Azure to azure-storage-blob's concurrent
+  downloader; GCP and OpenStack Swift use CloudBridge's generic driver, which
+  fetches ranges in parallel via cloned providers. A new
+  ``BucketObjectService.download_range`` primitive is also available directly
+  for partial reads. The per-call ``TransferConfig`` knobs (threshold, part
+  size, concurrency) now tune transfers in both directions — pass a different
+  instance per call to tune uploads and downloads differently.
+
 4.2.1 - July 10, 2026 (sha f06765d1100ec461908396475a4510460843a65c)
 --------------------------------------------------------------------
 
@@ -42,7 +59,7 @@ APIs it supported.
   underlying cloud SDKs are untyped. A ``mypy`` check runs in CI — the interface and base
   layers under a strict bar, and the SDK-wrapping providers under a pragmatic tier.
 * **Cross-provider multipart upload.** Large object uploads now use multipart transfer
-  with per-call ``UploadConfig`` knobs (threshold, part size, concurrency), with parts
+  with per-call ``TransferConfig`` knobs (threshold, part size, concurrency), with parts
   uploaded in parallel via cloned providers.
 
 ## Breaking changes

cloudbridge/base/resources.py

@@ -59,7 +59,7 @@
 from cloudbridge.interfaces.resources import SnapshotState
 from cloudbridge.interfaces.resources import Subnet
 from cloudbridge.interfaces.resources import SubnetState
-from cloudbridge.interfaces.resources import UploadConfig
+from cloudbridge.interfaces.resources import TransferConfig
 from cloudbridge.interfaces.resources import UploadPart
 from cloudbridge.interfaces.resources import VMFirewall
 from cloudbridge.interfaces.resources import VMFirewallRule
@@ -884,23 +884,107 @@ def save_content(self, target_stream: IO[bytes]) -> None:
         shutil.copyfileobj(
             cast("SupportsRead[bytes]", self.iter_content()), target_stream)
 
+    def download_to_file(self, path: str,
+                         config: TransferConfig | None = None) -> None:
+        size = self.size
+        if size <= self._multipart_threshold(config):
+            with open(path, 'wb') as f:
+                self.save_content(f)
+            return
+        self._download_ranged(path, size, config)
+
+    def _download_ranged(self, path: str, size: int,
+                         config: TransferConfig | None = None) -> None:
+        """
+        Fetch the object as ranged reads across a bounded thread pool,
+        writing each range at its offset into a preallocated file.
+
+        To stay safe even on providers whose SDK client/connection is not
+        thread-safe, each worker reads through its own cloned provider (see
+        :meth:`.CloudProvider.clone`), so no provider state is shared between
+        threads. Memory is bounded to ~concurrency * part_size. On any
+        failure the partial file is removed and the error re-raised.
+
+        Providers with an efficient, thread-safe native downloader (e.g. AWS
+        via boto3's ``download_file``, Azure via ``download_blob``) override
+        ``download_to_file`` to use it directly.
+        """
+        part_size = self._multipart_part_size(config)
+        if part_size < 1:
+            raise InvalidValueException('part_size', part_size)
+        concurrency = max(1, self._multipart_max_concurrency(config))
+        ranges = [(offset, min(part_size, size - offset))
+                  for offset in range(0, size, part_size)]
+        try:
+            with open(path, 'wb') as f:
+                f.truncate(size)
+            if concurrency == 1:
+                bucket_objects = self._bucket_objects
+                with open(path, 'r+b') as f:
+                    for offset, length in ranges:
+                        f.seek(offset)
+                        f.write(bucket_objects.download_range(
+                            self.bucket, self.name, offset, length))
+            else:
+                self._download_ranges_concurrently(path, ranges, concurrency)
+        except Exception:
+            try:
+                os.remove(path)
+            except OSError:
+                pass
+            raise
+
+    def _download_ranges_concurrently(
+            self, path: str, ranges: list[tuple[int, int]],
+            concurrency: int) -> None:
+        # A pool of cloned bucket-object services, one per worker, so each
+        # thread touches an isolated provider/connection.
+        clones: "queue.Queue[BucketObjectService]" = queue.Queue()
+        for _ in range(concurrency):
+            storage = cast("BaseStorageService",
+                           self._provider.clone().storage)
+            clones.put(storage._bucket_objects)
+
+        bucket = self.bucket
+        name = self.name
+
+        def fetch_one(offset: int, length: int) -> None:
+            service = clones.get()
+            try:
+                data = service.download_range(bucket, name, offset, length)
+            finally:
+                clones.put(service)
+            # Each worker writes through its own handle at its own offset;
+            # ranges never overlap, so no locking is needed. Data is released
+            # as soon as it is written, bounding memory to
+            # ~concurrency * part_size.
+            with open(path, 'r+b') as f:
+                f.seek(offset)
+                f.write(data)
+
+        with ThreadPoolExecutor(max_workers=concurrency) as executor:
+            futures = [executor.submit(fetch_one, offset, length)
+                       for offset, length in ranges]
+            for future in futures:
+                future.result()
+
     # The three resolvers below pick, in order of precedence: an explicit
-    # per-call UploadConfig field, the provider/global config, then the class
+    # per-call TransferConfig field, the provider/global config, then the class
     # default constant.
-    def _multipart_threshold(self, config: UploadConfig | None = None) -> int:
+    def _multipart_threshold(self, config: TransferConfig | None = None) -> int:
         if config is not None and config.threshold is not None:
             return int(config.threshold)
         return int(self._provider._get_config_value(
             'multipart_threshold', self.CB_MULTIPART_THRESHOLD))
 
-    def _multipart_part_size(self, config: UploadConfig | None = None) -> int:
+    def _multipart_part_size(self, config: TransferConfig | None = None) -> int:
         if config is not None and config.part_size is not None:
             return int(config.part_size)
         return int(self._provider._get_config_value(
             'multipart_part_size', self.CB_MULTIPART_PART_SIZE))
 
     def _multipart_max_concurrency(
-            self, config: UploadConfig | None = None) -> int:
+            self, config: TransferConfig | None = None) -> int:
         if config is not None and config.max_concurrency is not None:
             return int(config.max_concurrency)
         return int(self._provider._get_config_value(
@@ -936,22 +1020,22 @@ def _as_stream(data: str | bytes | IO[bytes]) -> IO[bytes]:
         return data
 
     def upload(self, data: str | bytes | IO[bytes],
-               config: UploadConfig | None = None) -> BucketObject:
+               config: TransferConfig | None = None) -> BucketObject:
         size = self._data_size(data)
         if size is not None and size > self._multipart_threshold(config):
             return self._upload_multipart(self._as_stream(data), config)
         return self._upload_single_shot(data)
 
     def upload_from_file(
             self, path: str,
-            config: UploadConfig | None = None) -> BucketObject:
+            config: TransferConfig | None = None) -> BucketObject:
         if os.path.getsize(path) > self._multipart_threshold(config):
             with open(path, 'rb') as f:
                 return self._upload_multipart(f, config)
         return self._upload_from_file_single_shot(path)
 
     def _upload_multipart(self, stream: IO[bytes],
-                          config: UploadConfig | None = None) -> BucketObject:
+                          config: TransferConfig | None = None) -> BucketObject:
         """
         Drive the explicit multipart lifecycle over a stream, reading it one
         part at a time so the whole payload is never held in memory.

cloudbridge/interfaces/resources.py

@@ -2161,37 +2161,43 @@ def delete(self) -> None:
         pass
 
 
-class UploadConfig(object):
+class TransferConfig(object):
     """
-    Provider-agnostic, per-call tuning for an object upload.
+    Provider-agnostic, per-call tuning for an object transfer in either
+    direction.
 
-    Passed optionally to :meth:`.BucketObject.upload` and
-    :meth:`.BucketObject.upload_from_file`. Any field left as ``None`` falls
-    back to the provider/global configuration (the ``CB_MULTIPART_*`` settings).
-    Each provider maps these fields onto its native transfer mechanism.
+    Passed optionally to :meth:`.BucketObject.upload`,
+    :meth:`.BucketObject.upload_from_file` and
+    :meth:`.BucketObject.download_to_file`. Any field left as ``None`` falls
+    back to the provider/global configuration (the ``CB_MULTIPART_*``
+    settings). Each provider maps these fields onto its native transfer
+    mechanism. To tune uploads and downloads differently, pass a different
+    instance to each call.
     """
 
     def __init__(self, threshold: int | None = None,
                  part_size: int | None = None,
                  max_concurrency: int | None = None) -> None:
         """
         :type threshold: ``int``
-        :param threshold: Size in bytes above which the upload is split into
-            multiple parts.
+        :param threshold: Size in bytes above which the transfer is split
+            into multiple parts.
 
         :type part_size: ``int``
-        :param part_size: Size in bytes of each part. Must be at least the
-            provider minimum (5 MiB on S3) for all but the final part.
+        :param part_size: Size in bytes of each part. For uploads this must
+            be at least the provider minimum (5 MiB on S3) for all but the
+            final part; downloads have no minimum.
 
         :type max_concurrency: ``int``
-        :param max_concurrency: Maximum number of parts to upload in parallel.
+        :param max_concurrency: Maximum number of parts to transfer in
+            parallel.
         """
         self.threshold = threshold
         self.part_size = part_size
         self.max_concurrency = max_concurrency
 
     def __repr__(self) -> str:
-        return ("<CB-UploadConfig: threshold={0}, part_size={1}, "
+        return ("<CB-TransferConfig: threshold={0}, part_size={1}, "
                 "max_concurrency={2}>".format(
                     self.threshold, self.part_size, self.max_concurrency))
 
@@ -2368,13 +2374,37 @@ def save_content(self, target_stream: IO[bytes]) -> None:
         """
         pass
 
+    @abstractmethod
+    def download_to_file(self, path: str,
+                         config: TransferConfig | None = None) -> None:
+        """
+        Download this object's content to a local file.
+
+        Objects larger than the configured transfer threshold are fetched as
+        ranged reads of ``part_size`` bytes, up to ``max_concurrency`` parts
+        in parallel, so large downloads are not bound to a single connection
+        (and the whole object is never held in memory). Smaller objects are
+        streamed in a single request. ``iter_content``/``save_content``
+        remain single-stream alternatives for arbitrary target streams.
+
+        :type path: ``str``
+        :param path: Local path to write the object's content to. An existing
+            file is overwritten; on failure no partial file is left behind.
+
+        :type config: :class:`.TransferConfig`
+        :param config: Optional per-call transfer tuning (threshold, part
+            size, concurrency). Any field left unset falls back to the
+            provider/global configuration.
+        """
+        pass
+
     @abstractmethod
     def upload(self, source_stream: IO[bytes],
-               config: UploadConfig | None = None) -> BucketObject | None:
+               config: TransferConfig | None = None) -> BucketObject | None:
         """
         Set the contents of the object to the data read from the source stream.
 
-        :type config: :class:`.UploadConfig`
+        :type config: :class:`.TransferConfig`
         :param config: Optional per-call upload tuning (multipart threshold,
             part size, concurrency). Any field left unset falls back to the
             provider/global configuration.
@@ -2386,7 +2416,7 @@ def upload(self, source_stream: IO[bytes],
 
     @abstractmethod
     def upload_from_file(self, path: str,
-                         config: UploadConfig | None = None) -> BucketObject | None:
+                         config: TransferConfig | None = None) -> BucketObject | None:
         """
         Store the contents of the file pointed by the "path" variable.
 
@@ -2396,7 +2426,7 @@ def upload_from_file(self, path: str,
         :type path: ``str``
         :param path: Absolute path to the file to be uploaded to S3.
 
-        :type config: :class:`.UploadConfig`
+        :type config: :class:`.TransferConfig`
         :param config: Optional per-call upload tuning (multipart threshold,
             part size, concurrency). Any field left unset falls back to the
             provider/global configuration.

cloudbridge/interfaces/services.py

@@ -1383,6 +1383,33 @@ def abort_multipart_upload(self, bucket: Bucket | str,
         """
         pass
 
+    @abstractmethod
+    def download_range(self, bucket: Bucket | str, object_name: str,
+                       offset: int, length: int) -> bytes:
+        """
+        Read a byte range of an object.
+
+        This is the primitive behind
+        :meth:`.BucketObject.download_to_file`'s parallel ranged downloads;
+        it is also usable directly for partial reads.
+
+        :type bucket: :class:`.Bucket`
+        :param bucket: The bucket containing the object.
+
+        :type object_name: ``str``
+        :param object_name: The key of the object to read.
+
+        :type offset: ``int``
+        :param offset: Zero-based byte offset to start reading at.
+
+        :type length: ``int``
+        :param length: Number of bytes to read.
+
+        :rtype: ``bytes``
+        :return: The requested byte range of the object's content.
+        """
+        pass
+
 
 class SecurityService(CloudService):
 

cloudbridge/providers/aws/resources.py

@@ -13,7 +13,7 @@
 from typing import TYPE_CHECKING
 from typing import cast
 
-from boto3.s3.transfer import TransferConfig
+from boto3.s3.transfer import TransferConfig as S3TransferConfig
 
 from botocore.exceptions import ClientError
 
@@ -58,7 +58,7 @@
 from cloudbridge.interfaces.resources import Subnet
 from cloudbridge.interfaces.resources import SubnetState
 from cloudbridge.interfaces.resources import TrafficDirection
-from cloudbridge.interfaces.resources import UploadConfig
+from cloudbridge.interfaces.resources import TransferConfig
 from cloudbridge.interfaces.resources import VMFirewall
 from cloudbridge.interfaces.resources import VMType
 from cloudbridge.interfaces.resources import Volume
@@ -920,30 +920,42 @@ def _upload_single_shot(self,
 
     def _upload_multipart(
             self, stream: IO[bytes],
-            config: UploadConfig | None = None) -> BucketObject:
+            config: TransferConfig | None = None) -> BucketObject:
         # boto3's TransferManager uploads parts concurrently with a thread-safe
         # client, so the transparent multipart path delegates to it rather than
         # CloudBridge's generic clone-pool driver.
-        transfer_config = TransferConfig(
+        transfer_config = S3TransferConfig(
             multipart_threshold=self._multipart_part_size(config),
             multipart_chunksize=self._multipart_part_size(config),
             max_concurrency=self._multipart_max_concurrency(config))
         self._obj.upload_fileobj(stream, Config=transfer_config)
         return self
 
     def upload_from_file(self, path: str,
-                         config: UploadConfig | None = None) -> BucketObject:
+                         config: TransferConfig | None = None) -> BucketObject:
         # boto3's upload_file streams large files in parts via its
         # TransferManager. Drive it with CloudBridge's multipart knobs so that
         # upload_from_file and upload() honour the same configuration rather
         # than boto3's own defaults.
-        transfer_config = TransferConfig(
+        transfer_config = S3TransferConfig(
             multipart_threshold=self._multipart_threshold(config),
             multipart_chunksize=self._multipart_part_size(config),
             max_concurrency=self._multipart_max_concurrency(config))
         self._obj.upload_file(path, Config=transfer_config)
         return self
 
+    def download_to_file(self, path: str,
+                         config: TransferConfig | None = None) -> None:
+        # boto3's TransferManager downloads large objects as parallel ranged
+        # GETs with a thread-safe client, so the transparent ranged path
+        # delegates to it rather than CloudBridge's generic clone-pool driver.
+        transfer_config = S3TransferConfig(
+            multipart_threshold=self._multipart_threshold(config),
+            multipart_chunksize=self._multipart_part_size(config),
+            max_concurrency=self._multipart_max_concurrency(config))
+        self._obj.meta.client.download_file(
+            self._obj.bucket_name, self.id, path, Config=transfer_config)
+
     def delete(self) -> None:
         self._obj.delete()