Add implementation of streaming abstractions

HISTORY.md

@@ -5,6 +5,22 @@
 - Added `AGENTS.md` with repository-specific guidance for coding agents covering contributor
   workflow, compatibility expectations, test rig and mock usage, live backend validation, and PR
   hygiene.
+- Added streaming I/O support for S3, Azure Blob Storage, Google Cloud Storage, and HTTP/HTTPS via `FileCacheMode.streaming`. (PR [#535](https://github.com/drivendataorg/cloudpathlib/pull/535))
+  - Added `FileCacheMode.streaming` enum value to enable direct streaming I/O without local caching.
+  - Added `CloudBufferedIO` class implementing `io.BufferedIOBase` for binary streaming operations.
+  - Added `CloudTextIO` class implementing `io.TextIOBase` for text streaming operations.
+  - Added provider-specific raw I/O implementations: `_S3StorageRaw`, `_AzureBlobStorageRaw`, `_GSStorageRaw`, `_HttpStorageRaw`.
+  - Added `register_raw_io_class` decorator for registering streaming I/O implementations.
+  - Added `buffer_size` parameter to `CloudPath.open()` for controlling streaming buffer size; the default is 5 MiB, matching the block sizes of comparable tools (a full-object `read()` always uses a single ranged request regardless of buffer size).
+  - Streaming upload extra args for S3 are filtered against botocore's bundled service model (also when the client object does not expose `meta`), so newly added S3 parameters are never silently dropped.
+  - Added a `streaming_max_concurrency` client parameter (default 1): each open streaming stream may issue up to that many requests in parallel — background part uploads while writing (in-flight memory bounded to concurrency × part size) and read-ahead prefetch of upcoming byte ranges while reading sequentially.
+  - Google Cloud Storage streaming writes use the XML API multipart upload (via the SDK's transfer-manager machinery) instead of a resumable-upload stream, matching the S3/Azure part mechanism and enabling concurrent part uploads.
+  - Streaming writes honor `force_overwrite_to_cloud` (and `CLOUDPATHLIB_FORCE_OVERWRITE_TO_CLOUD`), raising `OverwriteNewerCloudError` on close instead of overwriting an object that changed while the stream was open.
+  - `copy`/`rename`/`replace` work in streaming mode by streaming between clients instead of round-tripping through the local cache (`fspath`).
+  - Cache files created by the append/update fallback in streaming mode are cleaned up when the client is garbage collected.
+  - Streaming error paths raise `cloudpathlib.exceptions` types (`CloudPathFileNotFoundError`, `CloudPathNotImplementedError`), which subclass the corresponding builtins.
+- Changed `CloudPath.open(mode="a")` on a nonexistent cloud file to create it (matching the stdlib `open` and `pathlib`) instead of raising `CloudPathFileNotFoundError`. **Breaking change for users that relied on the previous error.** (PR [#535](https://github.com/drivendataorg/cloudpathlib/pull/535))
+- Changed the cached-write upload tie-break so a save that leaves the cache file's modification time exactly equal to the cloud version's (e.g., same-second writes on coarse-resolution filesystems) uploads instead of raising a spurious `OverwriteNewerCloudError`. (PR [#535](https://github.com/drivendataorg/cloudpathlib/pull/535))
 - Fixed mypy 2.x type errors in `Client` and `CloudPath` that caused CI lint failures (Issue [#563](https://github.com/drivendataorg/cloudpathlib/issues/563), PR [#566](https://github.com/drivendataorg/cloudpathlib/pull/566))
 - Changed `S3Client._get_metadata` to read object metadata with `HeadObject` instead of `GetObject`, so `stat`, `etag`, and `size` no longer open the object body. Also fixes a `KeyError` on `ContentLength` against S3-compatible gateways that drop `Content-Length` from `GetObject` responses. (Issue [#564](https://github.com/drivendataorg/cloudpathlib/issues/564), PR [#565](https://github.com/drivendataorg/cloudpathlib/pull/565))
 - Added a `lazy` keyword argument to `CloudPath.walk`. By default (`lazy=False`) the existing fast behavior is preserved: the whole subtree is fetched up front with a single recursive listing. Passing `lazy=True` lists each directory on demand so that, when `top_down=True`, callers can prune subdirectories by modifying `dirnames` in-place (à la `os.walk` / `Path.walk`) to skip fetching the contents of those subtrees entirely — dramatically reducing API calls for large, sparsely-traversed trees. (Issue [#518](https://github.com/drivendataorg/cloudpathlib/issues/518))

cloudpathlib/__init__.py

@@ -4,6 +4,7 @@
 from .anypath import AnyPath
 from .azure.azblobclient import AzureBlobClient
 from .azure.azblobpath import AzureBlobPath
+from .cloud_io import CloudBufferedIO, CloudTextIO
 from .cloudpath import CloudPath, implementation_registry
 from .patches import patch_open, patch_os_functions, patch_glob, patch_all_builtins
 from .gs.gsclient import GSClient
@@ -26,7 +27,9 @@
     "AnyPath",
     "AzureBlobClient",
     "AzureBlobPath",
+    "CloudBufferedIO",
     "CloudPath",
+    "CloudTextIO",
     "implementation_registry",
     "GSClient",
     "GSPath",

cloudpathlib/azure/__init__.py

@@ -1,5 +1,6 @@
 from .azblobclient import AzureBlobClient
 from .azblobpath import AzureBlobPath
+from .azure_io import _AzureBlobStorageRaw  # noqa: F401 - imported for registration
 
 __all__ = [
     "AzureBlobClient",

cloudpathlib/azure/azblobclient.py

@@ -3,18 +3,19 @@
 import os
 from http import HTTPStatus
 from pathlib import Path
-from typing import Any, Callable, Dict, Iterable, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Iterable, Optional, Sequence, Tuple, Union
 from itertools import islice
+from uuid import uuid4
 
 try:
     from typing import cast
 except ImportError:
     from typing_extensions import cast
 
-from ..client import Client, register_client_class
+from ..client import Client, _UploadPart, register_client_class
 from ..cloudpath import implementation_registry
 from ..enums import FileCacheMode
-from ..exceptions import MissingCredentialsError
+from ..exceptions import CloudPathFileNotFoundError, MissingCredentialsError
 from .azblobpath import AzureBlobPath
 
 try:
@@ -61,6 +62,7 @@ def __init__(
         file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
         local_cache_dir: Optional[Union[str, os.PathLike]] = None,
         content_type_method: Optional[Callable] = mimetypes.guess_type,
+        streaming_max_concurrency: int = 1,
     ):
         """Class constructor. Sets up a [`BlobServiceClient`](
         https://docs.microsoft.com/en-us/python/api/azure-storage-blob/azure.storage.blob.blobserviceclient?view=azure-python).
@@ -108,11 +110,15 @@ def __init__(
                 the `CLOUDPATHLIB_LOCAL_CACHE_DIR` environment variable.
             content_type_method (Optional[Callable]): Function to call to guess media type (mimetype) when
                 writing a file to the cloud. Defaults to `mimetypes.guess_type`. Must return a tuple (content type, content encoding).
+            streaming_max_concurrency (int): Maximum concurrent requests per open streaming
+                stream (background part uploads and read prefetch) when using
+                `FileCacheMode.streaming`; defaults to 1 (sequential).
         """
         super().__init__(
             local_cache_dir=local_cache_dir,
             content_type_method=content_type_method,
             file_cache_mode=file_cache_mode,
+            streaming_max_concurrency=streaming_max_concurrency,
         )
 
         if connection_string is None:
@@ -497,6 +503,92 @@ def _generate_presigned_url(
         url = f"{self._get_public_url(cloud_path)}?{sas_token}"
         return url
 
+    def _range_download(self, cloud_path: AzureBlobPath, start: int, end: int) -> bytes:
+        """Download a byte range from Azure Blob Storage."""
+        blob_client = self.service_client.get_blob_client(
+            container=cloud_path.container, blob=cloud_path.blob
+        )
+        try:
+            length = end - start + 1
+            downloader = blob_client.download_blob(offset=start, length=length)
+            return downloader.readall()
+        except ResourceNotFoundError:
+            raise CloudPathFileNotFoundError(f"Azure blob not found: {cloud_path}")
+        except HttpResponseError as e:
+            if (e.error and e.error.code == "InvalidRange") or e.status_code == 416:
+                return b""
+            raise
+
+    def _get_content_length(self, cloud_path: AzureBlobPath) -> int:
+        """Get the size of an Azure blob."""
+        blob_client = self.service_client.get_blob_client(
+            container=cloud_path.container, blob=cloud_path.blob
+        )
+        try:
+            properties = blob_client.get_blob_properties()
+            return properties.size
+        except ResourceNotFoundError:
+            raise CloudPathFileNotFoundError(f"Azure blob not found: {cloud_path}")
+
+    def _initiate_multipart_upload(self, cloud_path: AzureBlobPath) -> str:
+        """Return a unique session ID that namespaces this upload's block IDs.
+
+        Azure's uncommitted-block namespace is per-blob, so deterministic block IDs
+        would let concurrent writers to the same blob overwrite each other's staged
+        blocks and commit interleaved data.
+        """
+        return uuid4().hex
+
+    def _upload_part(
+        self, cloud_path: AzureBlobPath, upload_id: str, part_number: int, data: bytes
+    ) -> _UploadPart:
+        """Upload a block in an Azure block blob upload."""
+        import base64
+
+        blob_client = self.service_client.get_blob_client(
+            container=cloud_path.container, blob=cloud_path.blob
+        )
+        # Azure requires all block IDs for a blob to be the same length; uuid4().hex (32)
+        # plus a fixed-width part number keeps them uniform.
+        block_id = base64.b64encode(f"{upload_id}-{part_number:06d}".encode()).decode()
+        blob_client.stage_block(block_id=block_id, data=data, length=len(data))
+        return {"block_id": block_id}
+
+    def _complete_multipart_upload(
+        self, cloud_path: AzureBlobPath, upload_id: str, parts: Sequence[_UploadPart]
+    ) -> None:
+        """Commit an Azure block blob upload, threading content-type."""
+        blob_client = self.service_client.get_blob_client(
+            container=cloud_path.container, blob=cloud_path.blob
+        )
+        block_ids = [part["block_id"] for part in parts]
+        blob_client.commit_block_list(
+            block_ids, content_settings=self._streaming_content_settings(cloud_path)
+        )
+
+    def _streaming_content_settings(
+        self, cloud_path: AzureBlobPath
+    ) -> Optional["ContentSettings"]:
+        if self.content_type_method is None:
+            return None
+        content_type, content_encoding = self.content_type_method(str(cloud_path))
+        if not content_type and not content_encoding:
+            return None
+        return ContentSettings(content_type=content_type, content_encoding=content_encoding)
+
+    def _abort_multipart_upload(self, cloud_path: AzureBlobPath, upload_id: str) -> None:
+        """Let Azure expire uncommitted blocks."""
+        pass
+
+    def _put_empty_object(self, cloud_path: AzureBlobPath) -> None:
+        """Upload a zero-byte Azure blob, threading content-type."""
+        blob_client = self.service_client.get_blob_client(
+            container=cloud_path.container, blob=cloud_path.blob
+        )
+        blob_client.upload_blob(
+            b"", overwrite=True, content_settings=self._streaming_content_settings(cloud_path)
+        )
+
 
 def _hns_rmtree(data_lake_client, container, directory):
     """Stateless implementation so can be used in test suite cleanup as well.

cloudpathlib/azure/azure_io.py

@@ -0,0 +1,23 @@
+"""Azure Blob Storage streaming I/O."""
+
+from ..cloud_io import _CloudMultipartStorageRaw
+from ..cloudpath import register_raw_io_class
+
+
+@register_raw_io_class("azure")
+class _AzureBlobStorageRaw(_CloudMultipartStorageRaw):
+    """Azure range reads and block writes."""
+
+    # Azure permits at most 50,000 committed blocks.
+    _INITIAL_PART_SIZE = 4 * 1024 * 1024
+    _BLOCK_SIZE = _INITIAL_PART_SIZE
+    _MAX_PART_SIZE = 4_000 * 1024 * 1024
+    _MAX_BLOCK_SIZE = _MAX_PART_SIZE
+    _MAX_PARTS = 50_000
+    _BLOCKS_PER_SIZE_TIER = 1_000
+    _PARTS_PER_SIZE_TIER = _BLOCKS_PER_SIZE_TIER
+    _PROVIDER_NAME = "Azure block"
+
+    @classmethod
+    def _block_size_for_number(cls, block_number: int) -> int:
+        return cls._part_size_for_number(block_number)

cloudpathlib/client.py

@@ -4,13 +4,26 @@
 from pathlib import Path
 import shutil
 from tempfile import TemporaryDirectory
-from typing import ClassVar, Generic, Callable, Iterable, Optional, Tuple, TypeVar, Union
+from typing import (
+    Any,
+    Callable,
+    ClassVar,
+    Dict,
+    Generic,
+    Iterable,
+    Optional,
+    Sequence,
+    Tuple,
+    TypeVar,
+    Union,
+)
 
 from .cloudpath import CloudImplementation, CloudPath, implementation_registry
 from .enums import FileCacheMode
 from .exceptions import InvalidConfigurationException
 
 BoundedCloudPath = TypeVar("BoundedCloudPath", bound=CloudPath)
+_UploadPart = Dict[str, Any]
 
 
 def register_client_class(key: str) -> Callable:
@@ -34,11 +47,18 @@ def __init__(
         file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
         local_cache_dir: Optional[Union[str, os.PathLike]] = None,
         content_type_method: Optional[Callable] = mimetypes.guess_type,
-    ):
+        streaming_max_concurrency: int = 1,
+    ) -> None:
         self.file_cache_mode = None
         self._cache_tmp_dir = None
         self._cloud_meta.validate_completeness()
 
+        if streaming_max_concurrency < 1:
+            raise ValueError("streaming_max_concurrency must be at least 1")
+        # concurrent requests per open streaming stream (part uploads / read prefetch);
+        # 1 means fully sequential I/O
+        self.streaming_max_concurrency = streaming_max_concurrency
+
         # convert strings passed to enum
         if isinstance(file_cache_mode, str):
             file_cache_mode = FileCacheMode(file_cache_mode)
@@ -88,6 +108,9 @@ def __del__(self) -> None:
             FileCacheMode.tmp_dir,
             FileCacheMode.close_file,
             FileCacheMode.cloudpath_object,
+            # streaming avoids the cache except for append/update fallbacks, which
+            # should not outlive the client
+            FileCacheMode.streaming,
         ]:
             self.clear_cache()
 
@@ -184,3 +207,50 @@ def _generate_presigned_url(
         self, cloud_path: BoundedCloudPath, expire_seconds: int = 60 * 60
     ) -> str:
         pass
+
+    def _range_download(self, cloud_path: BoundedCloudPath, start: int, end: int) -> bytes:
+        """Download an inclusive byte range."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_range_download). "
+            "Implement this method or use a non-streaming file_cache_mode."
+        )
+
+    def _get_content_length(self, cloud_path: BoundedCloudPath) -> int:
+        """Return object size without downloading it."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_get_content_length)."
+        )
+
+    def _initiate_multipart_upload(self, cloud_path: BoundedCloudPath) -> str:
+        """Start a multipart upload."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_initiate_multipart_upload)."
+        )
+
+    def _upload_part(
+        self, cloud_path: BoundedCloudPath, upload_id: str, part_number: int, data: bytes
+    ) -> _UploadPart:
+        """Upload one part."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_upload_part)."
+        )
+
+    def _complete_multipart_upload(
+        self, cloud_path: BoundedCloudPath, upload_id: str, parts: Sequence[_UploadPart]
+    ) -> None:
+        """Complete a multipart upload."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_complete_multipart_upload)."
+        )
+
+    def _abort_multipart_upload(self, cloud_path: BoundedCloudPath, upload_id: str) -> None:
+        """Abort a multipart upload."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_abort_multipart_upload)."
+        )
+
+    def _put_empty_object(self, cloud_path: BoundedCloudPath) -> None:
+        """Create an empty object."""
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support streaming I/O (_put_empty_object)."
+        )