Add cross-provider ranged, parallel downloads

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 not 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 the generic driver, which
fetches ranges in parallel via cloned providers (memory bounded to
~concurrency * part_size; the partial file is removed on failure). A new
BucketObjectService.download_range primitive is also usable directly for
partial reads.

Since the same threshold/part-size/concurrency knobs now tune transfers
in both directions, UploadConfig (shipped only in 4.2.x, unused
downstream) is renamed to TransferConfig with no alias kept.

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 Snapshot
 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 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
         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,7 +1020,7 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
         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)
@@ -944,14 +1028,14 @@ class BaseBucketObject(BaseCloudResource, BucketObject):
 
     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,14 +2161,18 @@ class VMFirewallRule(CloudResource):
         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,
@@ -2176,22 +2180,24 @@ class UploadConfig(object):
                  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 @@ class BucketObject(CloudResource):
         """
         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 @@ class BucketObject(CloudResource):
 
     @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 @@ class BucketObject(CloudResource):
         :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 @@ class BucketObjectService(CloudService):
         """
         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 Iterator
 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 SnapshotState
 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,11 +920,11 @@ class AWSBucketObject(BaseBucketObject):
 
     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))
@@ -932,18 +932,30 @@ class AWSBucketObject(BaseBucketObject):
         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()
 

cloudbridge/providers/aws/services.py

@@ -733,6 +733,16 @@ class AWSBucketObjectService(BaseBucketObjectService):
             Bucket=cast(Any, bucket).name, Key=upload.object_name,
             UploadId=upload.id)
 
+    @dispatch(event="provider.storage._bucket_objects.download_range",
+              priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY)
+    def download_range(self, bucket: Bucket | str, object_name: str,
+                       offset: int, length: int) -> bytes:
+        client = cast("AWSCloudProvider", self.provider).s3_conn.meta.client
+        response = client.get_object(
+            Bucket=cast(Any, bucket).name, Key=object_name,
+            Range='bytes=%d-%d' % (offset, offset + length - 1))
+        return cast(bytes, response['Body'].read())
+
 
 class AWSComputeService(BaseComputeService):
 

cloudbridge/providers/azure/azure_client.py

@@ -507,6 +507,12 @@ class AzureClient(object):
         blob_client = self.blob_client(container_name, blob_name)
         blob_client.delete_blob(delete_snapshots)
 
+    def get_blob_range(self, container_name: str, blob_name: str,
+                       offset: int, length: int) -> bytes:
+        blob_client = self.blob_client(container_name, blob_name)
+        downloader = blob_client.download_blob(offset=offset, length=length)
+        return downloader.readall()
+
     def get_blob_url(self, container_name: Any, blob_name: str,
                      expiry_time: int, writable: bool,
                      content_disposition: str | None = None,

cloudbridge/providers/azure/resources.py

@@ -62,7 +62,7 @@ from cloudbridge.interfaces.resources import SnapshotState
 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
@@ -293,7 +293,7 @@ class AzureBucketObject(BaseBucketObject):
         return self
 
     def _upload_multipart(self, stream: IO[bytes],
-                          config: UploadConfig | None = None) -> BucketObject:
+                          config: TransferConfig | None = None) -> BucketObject:
         # The Azure SDK's upload_blob stages blocks concurrently (max_concurrency
         # workers) over a thread-safe client, so the transparent multipart path
         # delegates to it rather than CloudBridge's generic clone-pool driver.
@@ -302,6 +302,16 @@ class AzureBucketObject(BaseBucketObject):
             max_concurrency=self._multipart_max_concurrency(config))
         return self
 
+    def download_to_file(self, path: str,
+                         config: TransferConfig | None = None) -> None:
+        # azure-storage-blob's downloader fetches block ranges concurrently
+        # with a thread-safe client, so delegate to it rather than
+        # CloudBridge's generic clone-pool driver.
+        with open(path, 'wb') as f:
+            self._blob_client.download_blob(
+                max_concurrency=self._multipart_max_concurrency(config)
+            ).readinto(f)
+
     def delete(self) -> None:
         """
         Delete this object.

cloudbridge/providers/azure/services.py

@@ -766,6 +766,14 @@ class AzureBucketObjectService(BaseBucketObjectService):
                   "%s/%s will expire automatically.",
                   cast("Bucket", bucket).name, upload.object_name)
 
+    @dispatch(event="provider.storage._bucket_objects.download_range",
+              priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY)
+    def download_range(self, bucket: Bucket | str, object_name: str,
+                       offset: int, length: int) -> bytes:
+        azure_client = cast("AzureCloudProvider", self.provider).azure_client
+        return cast(bytes, azure_client.get_blob_range(
+            cast("Bucket", bucket).name, object_name, offset, length))
+
 
 class AzureComputeService(BaseComputeService):
     def __init__(self, provider: CloudProvider) -> None:

cloudbridge/providers/gcp/resources.py

@@ -81,7 +81,7 @@ if TYPE_CHECKING:
     from cloudbridge.interfaces.resources import Snapshot
     from cloudbridge.interfaces.resources import Subnet
     from cloudbridge.interfaces.resources import SubnetSubService
-    from cloudbridge.interfaces.resources import UploadConfig
+    from cloudbridge.interfaces.resources import TransferConfig
     from cloudbridge.interfaces.resources import VMFirewall
     from cloudbridge.interfaces.resources import VMFirewallRuleSubService
     from cloudbridge.interfaces.resources import VMType
@@ -2169,7 +2169,7 @@ class GCPBucketObject(BaseBucketObject):
         return self
 
     def upload_from_file(self, path: str,
-                         config: UploadConfig | None = None) -> BucketObject:
+                         config: TransferConfig | None = None) -> BucketObject:
         """
         Upload a binary file.
 

cloudbridge/providers/gcp/services.py

@@ -1769,6 +1769,17 @@ class GCPBucketObjectService(BaseBucketObjectService):
         self._delete_objects(bucket, self._list_temp_objects(bucket, upload),
                              ignore_missing=True)
 
+    @dispatch(event="provider.storage._bucket_objects.download_range",
+              priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY)
+    def download_range(self, bucket: Bucket | str, object_name: str,
+                       offset: int, length: int) -> bytes:
+        provider = cast("GCPCloudProvider", self.provider)
+        request = provider.gcp_storage.objects().get_media(
+            bucket=cast(Any, bucket).name, object=object_name)
+        request.headers['range'] = 'bytes=%d-%d' % (
+            offset, offset + length - 1)
+        return cast(bytes, request.execute())
+
     def _compose(self, bucket: Bucket | str, upload: MultipartUpload,
                  destination: str, sources: builtins.list[str]) -> builtins.list[str]:
         """

cloudbridge/providers/openstack/resources.py

@@ -64,7 +64,7 @@ from cloudbridge.interfaces.resources import SnapshotState
 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
@@ -1413,7 +1413,7 @@ class OpenStackBucketObject(BaseBucketObject):
         return self
 
     def upload_from_file(self, path: str,
-                         config: UploadConfig | None = None) -> BucketObject:
+                         config: TransferConfig | None = None) -> BucketObject:
         """
         Stores the contents of the file pointed by the ``path`` variable.
         If the file is bigger than 5 Gig, it will be broken into segments.

cloudbridge/providers/openstack/services.py

@@ -808,6 +808,16 @@ class OpenStackBucketObjectService(BaseBucketObjectService):
             except SwiftClientException:
                 pass  # idempotent: ignore already-deleted segments
 
+    @dispatch(event="provider.storage._bucket_objects.download_range",
+              priority=BaseBucketObjectService.STANDARD_EVENT_PRIORITY)
+    def download_range(self, bucket: Bucket | str, object_name: str,
+                       offset: int, length: int) -> bytes:
+        provider = cast("OpenStackCloudProvider", self.provider)
+        _, data = provider.swift.get_object(
+            cast(Any, bucket).name, object_name,
+            headers={'Range': 'bytes=%d-%d' % (offset, offset + length - 1)})
+        return cast(bytes, data)
+
 
 class OpenStackComputeService(BaseComputeService):
 

tests/test_download_driver.py

@@ -0,0 +1,236 @@
+"""
+Provider-agnostic unit tests for the base ranged download driver
+(``BaseBucketObject.download_to_file`` / ``_download_ranged``).
+
+The driver is the engine behind transparent large downloads on providers that
+do not override it (GCP, OpenStack Swift). Because the mock provider is
+AWS-backed and AWS overrides the driver with boto3's native downloader, the
+driver is exercised here directly against in-memory fakes so it has coverage
+in CI without cloud credentials.
+"""
+import os
+import tempfile
+import threading
+import unittest
+
+from cloudbridge.base.resources import BaseBucketObject
+from cloudbridge.interfaces.exceptions import InvalidValueException
+from cloudbridge.interfaces.resources import TransferConfig
+
+
+class _Recorder:
+    """Thread-safe range log shared by the original and cloned fake
+    services."""
+
+    def __init__(self, content):
+        self.content = content
+        self._lock = threading.Lock()
+        self.ranges = []            # (offset, length) served
+        self.services_used = set()  # id() of each service that served a range
+        self.clone_count = 0
+        self.single_shot = False
+        self.active = 0
+        self.max_active = 0
+        self.fail_on_offset = None  # offset that should raise
+
+    def serve_range(self, service, offset, length):
+        with self._lock:
+            self.active += 1
+            self.max_active = max(self.max_active, self.active)
+        try:
+            if self.fail_on_offset == offset:
+                raise RuntimeError("boom at offset %d" % offset)
+            # Hold briefly so concurrent fetches genuinely overlap.
+            threading.Event().wait(0.02)
+            with self._lock:
+                self.ranges.append((offset, length))
+                self.services_used.add(id(service))
+            return self.content[offset:offset + length]
+        finally:
+            with self._lock:
+                self.active -= 1
+
+
+class _FakeService:
+    def __init__(self, recorder, provider):
+        self._recorder = recorder
+        self._provider = provider
+
+    def download_range(self, bucket, object_name, offset, length):
+        return self._recorder.serve_range(self, offset, length)
+
+
+class _FakeStorage:
+    def __init__(self, service):
+        self._bucket_objects = service
+
+
+class _FakeProvider:
+    def __init__(self, recorder):
+        self._recorder = recorder
+        self.storage = _FakeStorage(_FakeService(recorder, self))
+
+    def clone(self, zone=None):
+        self._recorder.clone_count += 1
+        return _FakeProvider(self._recorder)
+
+    def _get_config_value(self, key, default_value=None):
+        return default_value
+
+
+class _DriverObject(BaseBucketObject):
+    """A BaseBucketObject wired to fakes with tiny transfer sizes."""
+
+    def __init__(self, provider, threshold, part_size, concurrency):
+        super(_DriverObject, self).__init__(provider)
+        self._threshold = threshold
+        self._part_size = part_size
+        self._concurrency = concurrency
+
+    @property
+    def id(self):
+        return "obj"
+
+    @property
+    def name(self):
+        return "obj"
+
+    @property
+    def size(self):
+        return len(self._provider._recorder.content)
+
+    @property
+    def bucket(self):
+        return "BUCKET"
+
+    def save_content(self, target_stream):
+        self._provider._recorder.single_shot = True
+        target_stream.write(self._provider._recorder.content)
+
+    def _multipart_threshold(self, config=None):
+        if config is not None and config.threshold is not None:
+            return config.threshold
+        return self._threshold
+
+    def _multipart_part_size(self, config=None):
+        if config is not None and config.part_size is not None:
+            return config.part_size
+        return self._part_size
+
+    def _multipart_max_concurrency(self, config=None):
+        if config is not None and config.max_concurrency is not None:
+            return config.max_concurrency
+        return self._concurrency
+
+
+class DownloadDriverTestCase(unittest.TestCase):
+
+    def _driver(self, recorder, threshold, part_size, concurrency):
+        return _DriverObject(
+            _FakeProvider(recorder), threshold, part_size, concurrency)
+
+    def _download(self, driver, config=None):
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        os.remove(path)
+        try:
+            driver.download_to_file(path, config)
+            with open(path, 'rb') as f:
+                return f.read()
+        finally:
+            if os.path.exists(path):
+                os.remove(path)
+
+    def test_reassembles_content_in_order(self):
+        content = b"abcdefghijABCDEFGHIJ0123456789x"  # 31 bytes -> 8 ranges
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=10, part_size=4, concurrency=3)
+        self.assertEqual(self._download(driver), content)
+        self.assertFalse(recorder.single_shot)
+        # Ranges tile the object exactly: no gaps, no overlap, short tail.
+        self.assertEqual(
+            sorted(recorder.ranges),
+            [(offset, min(4, 31 - offset)) for offset in range(0, 31, 4)])
+
+    def test_below_threshold_uses_single_shot(self):
+        content = b"tiny content"
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=100, part_size=4, concurrency=3)
+        self.assertEqual(self._download(driver), content)
+        self.assertTrue(recorder.single_shot)
+        self.assertEqual(recorder.ranges, [])
+
+    def test_downloads_ranges_concurrently_via_cloned_services(self):
+        concurrency = 4
+        content = bytes(range(12))  # 12 ranges of one byte each
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=1, part_size=1, concurrency=concurrency)
+        self.assertEqual(self._download(driver), content)
+
+        # A clone per worker, reused across ranges.
+        self.assertEqual(recorder.clone_count, concurrency)
+        self.assertEqual(len(recorder.services_used), concurrency)
+        # Real parallelism happened, bounded by the configured concurrency.
+        self.assertGreater(recorder.max_active, 1)
+        self.assertLessEqual(recorder.max_active, concurrency)
+
+    def test_single_concurrency_does_not_clone(self):
+        content = b"abcdefghij"
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=1, part_size=4, concurrency=1)
+        self.assertEqual(self._download(driver), content)
+        self.assertEqual(recorder.clone_count, 0)
+        self.assertEqual(recorder.max_active, 1)
+
+    def test_per_call_config_overrides_concurrency(self):
+        content = bytes(range(12))
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=1, part_size=1, concurrency=1)
+        result = None
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        try:
+            driver.download_to_file(path, TransferConfig(max_concurrency=3))
+            with open(path, 'rb') as f:
+                result = f.read()
+        finally:
+            os.remove(path)
+        self.assertEqual(result, content)
+        self.assertEqual(recorder.clone_count, 3)
+        self.assertGreater(recorder.max_active, 1)
+        self.assertLessEqual(recorder.max_active, 3)
+
+    def test_removes_partial_file_and_raises_on_range_failure(self):
+        content = bytes(range(16))
+        recorder = _Recorder(content)
+        recorder.fail_on_offset = 8
+        driver = self._driver(
+            recorder, threshold=1, part_size=4, concurrency=2)
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        os.remove(path)
+        try:
+            with self.assertRaises(Exception):
+                driver.download_to_file(path)
+            self.assertFalse(os.path.exists(path))
+        finally:
+            if os.path.exists(path):
+                os.remove(path)
+
+    def test_part_size_must_be_positive(self):
+        content = bytes(range(16))
+        recorder = _Recorder(content)
+        driver = self._driver(
+            recorder, threshold=1, part_size=4, concurrency=2)
+        with self.assertRaises(InvalidValueException):
+            self._download(driver, TransferConfig(part_size=0))
+        self.assertEqual(recorder.ranges, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_multipart_driver.py

@@ -16,7 +16,7 @@ from cloudbridge.base.resources import BaseBucketObject
 from cloudbridge.base.resources import BaseMultipartUpload
 from cloudbridge.base.resources import BaseUploadPart
 from cloudbridge.interfaces.exceptions import InvalidValueException
-from cloudbridge.interfaces.resources import UploadConfig
+from cloudbridge.interfaces.resources import TransferConfig
 
 
 class _Recorder:
@@ -193,7 +193,7 @@ class MultipartDriverTestCase(unittest.TestCase):
         # bumps it to 3, engaging the clone-per-worker pool.
         driver = self._driver(recorder, part_size=1, concurrency=1)
         driver._upload_multipart(BytesIO(b"0123456789ab"),
-                                 UploadConfig(max_concurrency=3))
+                                 TransferConfig(max_concurrency=3))
         self.assertEqual(recorder.clone_count, 3)
         self.assertGreater(recorder.max_active, 1)
         self.assertLessEqual(recorder.max_active, 3)

tests/test_object_store_service.py

@@ -14,7 +14,7 @@ from cloudbridge.interfaces.exceptions import DuplicateResourceException
 from cloudbridge.interfaces.provider import TestMockHelperMixin
 from cloudbridge.interfaces.resources import Bucket
 from cloudbridge.interfaces.resources import BucketObject
-from cloudbridge.interfaces.resources import UploadConfig
+from cloudbridge.interfaces.resources import TransferConfig
 
 from tests import helpers
 from tests.helpers import ProviderTestBase
@@ -452,7 +452,7 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
         # knobs (not boto3's own defaults). This wiring is AWS-specific.
         if self.provider.PROVIDER_ID not in ('aws', 'mock'):
             self.skipTest("TransferConfig wiring is specific to AWS")
-        from boto3.s3.transfer import TransferConfig
+        from boto3.s3.transfer import TransferConfig as S3TransferConfig
 
         name = "cbtest-mpu-{0}".format(helpers.get_uuid())
         test_bucket = self.provider.storage.buckets.create(name)
@@ -476,16 +476,16 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
 
                 spy.assert_called_once()
                 config = spy.call_args.kwargs['Config']
-                self.assertIsInstance(config, TransferConfig)
+                self.assertIsInstance(config, S3TransferConfig)
                 self.assertEqual(config.multipart_chunksize, 7 * 1024 * 1024)
                 self.assertEqual(config.max_concurrency, 3)
 
     @helpers.skipIfNoService(['storage.buckets'])
     def test_per_call_upload_config_overrides_defaults(self):
-        # A per-call UploadConfig takes precedence over the global knobs.
+        # A per-call TransferConfig takes precedence over the global knobs.
         if self.provider.PROVIDER_ID not in ('aws', 'mock'):
             self.skipTest("TransferConfig wiring is specific to AWS")
-        from boto3.s3.transfer import TransferConfig
+        from boto3.s3.transfer import TransferConfig as S3TransferConfig
 
         name = "cbtest-mpu-{0}".format(helpers.get_uuid())
         test_bucket = self.provider.storage.buckets.create(name)
@@ -496,8 +496,8 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
             with cb_helpers.cleanup_action(lambda: obj.delete()):
                 test_file = os.path.join(
                     helpers.get_test_fixtures_folder(), 'logo.jpg')
-                cfg = UploadConfig(part_size=8 * 1024 * 1024,
-                                   max_concurrency=2)
+                cfg = TransferConfig(part_size=8 * 1024 * 1024,
+                                     max_concurrency=2)
                 # pylint:disable=protected-access
                 with mock.patch.object(
                         obj._obj, 'upload_file',
@@ -505,10 +505,99 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
                     obj.upload_from_file(test_file, config=cfg)
 
                 config = spy.call_args.kwargs['Config']
-                self.assertIsInstance(config, TransferConfig)
+                self.assertIsInstance(config, S3TransferConfig)
                 self.assertEqual(config.multipart_chunksize, 8 * 1024 * 1024)
                 self.assertEqual(config.max_concurrency, 2)
 
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_download_to_file_roundtrip(self):
+        name = "cbtest-dl-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("roundtrip.txt")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                content = b"a small payload below the download threshold"
+                obj.upload(content)
+
+                stored = test_bucket.objects.get("roundtrip.txt")
+                fd, path = tempfile.mkstemp()
+                os.close(fd)
+                with cb_helpers.cleanup_action(lambda: os.remove(path)):
+                    stored.download_to_file(path)
+                    with open(path, 'rb') as f:
+                        self.assertEqual(f.read(), content)
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_transparent_download_large_uses_ranged(self):
+        name = "cbtest-dl-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("ranged.bin")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                # Distinct byte pattern so out-of-order reassembly would show.
+                content = bytes(range(256)) * ((MIN_PART_SIZE * 2) // 256)
+                obj.upload(BytesIO(content),
+                           TransferConfig(threshold=MIN_PART_SIZE,
+                                          part_size=MIN_PART_SIZE))
+
+                stored = test_bucket.objects.get("ranged.bin")
+                fd, path = tempfile.mkstemp()
+                os.close(fd)
+                with cb_helpers.cleanup_action(lambda: os.remove(path)):
+                    # Lower the threshold so the download crosses it (each
+                    # provider routes its own way underneath) and assert the
+                    # content round-trips exactly.
+                    stored.download_to_file(
+                        path, TransferConfig(threshold=MIN_PART_SIZE,
+                                             part_size=MIN_PART_SIZE,
+                                             max_concurrency=2))
+                    with open(path, 'rb') as f:
+                        self.assertEqual(f.read(), content)
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_download_to_file_uses_transfer_config(self):
+        # AWS drives boto3's TransferManager with CloudBridge's transfer
+        # knobs (not boto3's own defaults). This wiring is AWS-specific.
+        if self.provider.PROVIDER_ID not in ('aws', 'mock'):
+            self.skipTest("TransferConfig wiring is specific to AWS")
+        from boto3.s3.transfer import TransferConfig as S3TransferConfig
+
+        name = "cbtest-dl-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj = test_bucket.objects.create("dlconfig.bin")
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                obj.upload(b"some downloadable content")
+                stored = test_bucket.objects.get("dlconfig.bin")
+
+                fd, path = tempfile.mkstemp()
+                os.close(fd)
+                with cb_helpers.cleanup_action(lambda: os.remove(path)):
+                    cfg = TransferConfig(part_size=8 * 1024 * 1024,
+                                         max_concurrency=2)
+                    # pylint:disable=protected-access
+                    with mock.patch.object(
+                            stored._obj.meta.client, 'download_file',
+                            wraps=stored._obj.meta.client
+                            .download_file) as spy:
+                        stored.download_to_file(path, config=cfg)
+
+                    spy.assert_called_once()
+                    config = spy.call_args.kwargs['Config']
+                    self.assertIsInstance(config, S3TransferConfig)
+                    self.assertEqual(
+                        config.multipart_chunksize, 8 * 1024 * 1024)
+                    self.assertEqual(config.max_concurrency, 2)
+                    with open(path, 'rb') as f:
+                        self.assertEqual(
+                            f.read(), b"some downloadable content")
+
     @skip("Skip unless you want to test objects bigger than 5GB")
     @helpers.skipIfNoService(['storage.buckets'])
     def test_upload_download_bucket_content_with_large_file(self):