Repository navigation
Add implementation of streaming abstractions - #535
Merged
Merged
Conversation
Contributor
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #535 +/- ##
========================================
+ Coverage 95.1% 96.1% +0.9%
========================================
Files 28 30 +2
Lines 2286 2977 +691
========================================
+ Hits 2176 2861 +685
- Misses 110 116 +6
🚀 New features to boost your workflow:
|
Contributor
There was a problem hiding this comment.
Pull Request Overview
This PR adds streaming I/O support to cloudpathlib, enabling direct file streaming from/to cloud storage (S3, Azure Blob Storage, GCS, and HTTP/HTTPS) without local caching. Key changes include:
- New
FileCacheMode.streamingenum value for direct streaming without caching CloudBufferedIOandCloudTextIOclasses implementing standard Python I/O interfaces- Provider-specific raw I/O adapters for efficient range requests and multipart uploads
Reviewed Changes
Copilot reviewed 32 out of 32 changed files in this pull request and generated 28 comments.
Show a summary per file
| File | Description |
|---|---|
cloudpathlib/enums.py |
Added streaming enum value to FileCacheMode |
cloudpathlib/cloud_io.py |
New module with CloudBufferedIO, CloudTextIO, and _CloudStorageRaw base classes |
cloudpathlib/cloudpath.py |
Updated CloudImplementation to track raw I/O classes; modified open() and __fspath__() for streaming mode |
cloudpathlib/client.py |
Added abstract methods for streaming I/O operations |
cloudpathlib/s3/s3_io.py |
S3-specific streaming I/O implementation |
cloudpathlib/azure/azure_io.py |
Azure-specific streaming I/O implementation |
cloudpathlib/gs/gs_io.py |
GCS-specific streaming I/O implementation |
cloudpathlib/http/http_io.py |
HTTP-specific streaming I/O implementation |
tests/test_cloud_io.py |
Comprehensive test suite for streaming I/O functionality |
tests/conftest.py |
Updated test fixtures to use CloudImplementation with raw I/O classes |
tests/test_cloudpath_instantiation.py |
Updated test to access dependencies_loaded via cloud_implementation |
tests/mock_clients/mock_s3.py |
Added mock methods for range requests and multipart uploads |
tests/mock_clients/mock_gs.py |
Added mock methods for range downloads and uploads |
tests/mock_clients/mock_azureblob.py |
Added mock methods for range downloads and block uploads |
tests/http_fixtures.py |
Added Range request handling to HTTP test server |
docs/docs/streaming_io.md |
Comprehensive documentation for streaming I/O feature |
docs/docs/caching.ipynb |
Updated caching documentation to include streaming mode |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| rig = CloudProviderTestRig( | ||
| path_class=LocalS3Path, | ||
| client_class=LocalS3Client, | ||
| cloud_implementation=local_s3_implementation, |
There was a problem hiding this comment.
Incorrect variable used: local_s3_implementation should be local_s3_cloud_implementation. This passes the wrong CloudImplementation object (the one from local.implementations.s3 module instead of the test-specific one created on line 559).
Suggested change
| cloud_implementation=local_s3_implementation, | |
| cloud_implementation=local_s3_cloud_implementation, |
|
|
||
| def validate_completeness(self) -> None: | ||
| expected = ["client_class", "path_class"] | ||
| expected = ["client_class", "path_class", "raw_io_class"] |
There was a problem hiding this comment.
The validation now requires raw_io_class to be present, but this is a breaking change. Existing custom implementations won't have this attribute set, causing them to fail validation. Consider making raw_io_class optional for backward compatibility, or only validate it when streaming mode is enabled.
Suggested change
| expected = ["client_class", "path_class", "raw_io_class"] | |
| expected = ["client_class", "path_class"] |
| # Flush the buffer first (which will call write() on the raw stream) | ||
| try: | ||
| self._buffer.flush() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| # Try to access the internal _closed attribute | ||
| if hasattr(self._buffer, "_closed"): | ||
| object.__setattr__(self._buffer, "_closed", True) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
Suggested change
| except Exception: | |
| except Exception: | |
| # Ignore errors when setting _closed; not critical for buffer cleanup. |
| path.client.file_cache_mode = original_mode | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| finally: | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| finally: | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| finally: | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| finally: | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| finally: | ||
| try: | ||
| path.unlink() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
Introduces a new FileCacheMode.streaming that opens cloud files as
Python-standard BufferedIOBase/TextIOWrapper objects backed by provider
range-requests and multipart/resumable uploads, without requiring a local
cache copy.
New public API:
- CloudBufferedIO: BufferedIOBase wrapper for binary streaming
- CloudTextIO: TextIOWrapper for text streaming
- FileCacheMode.streaming enum value
Provider implementations:
- S3: multipart upload with 5 MiB minimum non-final part size;
_put_empty_object for zero-byte files
- GCS: blob.open("wb") resumable write stream; _put_empty_object
- Azure Blob: block-blob stage/commit pipeline; _put_empty_object
- HTTP(S): single-PUT via self.opener (honours SSL context)
- LocalClient: _LocalWriteStream + UUID-keyed multipart buffers for
concurrent-write safety
Correctness fixes included:
- _CloudStorageRaw.close() propagates finalize errors via try/finally
- seekable() returns False for write-only streams
- Exclusive-create ("x") mode checked against cloud existence
- Append/update modes ("a", "+") fall back to file-cache path
- Client streaming methods default to NotImplementedError (not
@AbstractMethod) so custom Client subclasses don't require
re-implementation
- HTTP _put_data uses self.opener for SSL-aware PUT requests
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- test_s3_multipart_upload, test_azure_block_upload, test_gs_resumable_upload were skipping on ALL rigs because hasattr(rig, "s3_path") etc. is always False; switch to rig.path_class.cloud_prefix comparisons so they run on the correct rigs - gs/gsclient.py GCSNotFound fallback: type: ignore[misc, assignment] - azure/azblobclient.py: e.error_code → e.error.code (guarded for None) - gs/gs_io.py: annotate _writer as Optional[Any] to satisfy mypy Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Review fixes (verified by adversarial review of this PR): - write() advances the raw position so tell() is correct on streaming write streams (fixes corrupt output from zipfile and other position-dependent writers) - seek() on write-only raw streams raises io.UnsupportedOperation as documented instead of silently accepting the seek - GS range errors are matched structurally (status 416) instead of by a "416" substring that could swallow unrelated transient errors - Azure block IDs are namespaced by a per-upload session ID so concurrent writers cannot clobber each other's staged blocks - read() to EOF fetches the remainder in one ranged request (readall) instead of one request per 8 KiB - copy()/rename()/replace() work in streaming mode by streaming between clients instead of round-tripping through fspath - streaming writes honor force_overwrite_to_cloud, raising OverwriteNewerCloudError on close instead of clobbering a newer object - Client.__del__ cleans up streaming-mode cache files from the append/update fallback; docs no longer claim streaming never touches disk - streaming error paths raise cloudpathlib exception types; failed size lookups are memoized per stream; negative buffering matches builtins.open; behavior changes recorded in HISTORY.md Coverage and fidelity: - S3 extra-args filtering consults botocore's bundled service model instead of a hand-copied (and already stale) fallback table, with a drift-guard test - mocks now raise realistic SDK errors for missing objects and past-EOF ranges; HTTP test server clamps range ends per RFC 7233; GS mock enforces the 5 MiB non-final part minimum - default streaming buffer raised to 5 MiB; regression tests for all of the above plus parquet-over-streaming compatibility (pyarrow dev dep) Streaming concurrency: - new streaming_max_concurrency client parameter: bounded background part uploads while writing and read-ahead prefetch while reading, sequential by default - GS streaming writes use the XML API multipart upload (via the SDK's transfer-manager machinery) with in-memory parts, replacing the resumable write stream and unifying all multipart providers on one write contract; mock GS implements the XML MPU wire protocol Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Streaming moves can retain their source, conflict checks remain racy, and large writes can exceed the documented memory bound.
Review effort: Balanced
Findings: 2
Open (7)
Comment on lines
+221
to
+223
| if self._pre_finalize is not None: | ||
| self._pre_finalize() | ||
| self._finalize_upload() |
Comment on lines
+1393
to
+1399
| with self.open("rb") as src_file: | ||
| with target_path.open("wb", force_overwrite_to_cloud=True) as dst_file: | ||
| # the streaming buffer size rather than shutil's 64 KiB default, so | ||
| # each buffered read is forwarded in one iteration | ||
| shutil.copyfileobj(src_file, dst_file, env.streaming_buffer_size()) | ||
|
|
||
| return cast(Union[Path, Self], target_path) |
Comment on lines
+80
to
+82
| if streaming_max_concurrency < 1: | ||
| raise ValueError("streaming_max_concurrency must be at least 1") | ||
| self.streaming_max_concurrency = streaming_max_concurrency |
Comment on lines
+429
to
+437
| def _upload_chunk(self, data: _ReadableBuffer) -> None: | ||
| view = memoryview(data) | ||
| if view.nbytes == 0: | ||
| return | ||
| self._write_buffer.extend(view) | ||
| target_size = self._target_part_size() | ||
| while len(self._write_buffer) >= target_size: | ||
| self._upload_buffered_part(target_size) | ||
| target_size = self._target_part_size() |
Comment on lines
+427
to
+430
| body = response["Body"] | ||
| data = body.read() | ||
| body.close() | ||
| return data |
Comment on lines
+770
to
+772
| "execution_count": null, | ||
| "metadata": {}, | ||
| "outputs": [], |
Comment on lines
+7
to
+8
| Nothing is written to disk, only the part of the object you read is downloaded, and | ||
| written data is uploaded while you write it. |
…ation - move() across clients now removes the source after the destination is written, in both the streaming branch and the cached branch (which never did on master). - A single write larger than the part size is sliced into parts as it is copied, so the part buffer never exceeds one part. - streaming_max_concurrency must be an int (not a float or bool). - S3 ranged reads close the response body even when the read fails. - Docs no longer claim streaming never touches disk (HTTP uploads spool). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The two streaming example cells had never been executed and the script export was stale. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The fallbacks run at import time, so the tests re-import gsclient in a subprocess with the relevant modules made unimportable (reloading in-process would re-register GSClient and desynchronise the rest of the suite). Coverage's subprocess patch is enabled so those lines count. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The inline-code part of the 'File cache mode' headers was dropped from the rendered table of contents, leaving five identical entries. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This branch was successfully deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds streaming I/O: a new
FileCacheMode.streamingthat makesCloudPath.open()read and write cloud objects directly instead of going through the local file cache.What it does
client = S3Client(file_cache_mode=FileCacheMode.streaming)(orCLOUDPATHLIB_FILE_CACHE_MODE=streaming). Thenpath.open(...)returns the standardio.BufferedReader,io.BufferedWriter, orio.TextIOWrapper, backed by the cloud object.PUT; larger ones use multipart uploads (S3, GCS XML API) or block uploads (Azure). The object only appears when the stream is closed; a failed write aborts the upload and leaves no partial object.cloudpathlib.localmock clients.bufferinghas the same meaning as in the builtinopen(). The default buffer (5 MiB) is also the size of each ranged request.streaming_max_concurrencysetting (default 4) that bounds background read-ahead on sequential reads and background part uploads on writes.force_overwrite_to_cloudworks as in cached mode: closing a write stream raisesOverwriteNewerCloudErrorif the object changed while the stream was open.xmodes raiseCloudPathFileExistsErrorif the object appeared in the meantime.copy,rename, andreplacebetween clients stream from one object to the other.CLOUDPATHLIB_STREAMING_BUFFER_SIZE,CLOUDPATHLIB_STREAMING_MAX_CONCURRENCY,CLOUDPATHLIB_STREAMING_PART_SIZE, andCLOUDPATHLIB_<S3|AZURE|GS>_STREAMING_PART_SIZE.Limitations
a,r+,w+, ...) cannot stream over object storage. They fall back to the local cache (download, modify, re-upload on close, then remove the cache file) and emit aUserWarning.fspathraisesCloudPathNotImplementedErrorin streaming mode because there is no local file. Pass the open file object to libraries instead of a path.Rangerequests. HTTP writes are a single request on close.cloudpathlib.exceptionstypes;CloudPathStreamingError(anOSError) is new.Implementation
cloudpathlib/cloud_io.py: aRawIOBaseadapter over client hooks (_range_download,_put_object,_initiate_multipart_upload,_upload_part,_complete_multipart_upload,_abort_multipart_upload), wrapped by the stdlib buffered and text classes. EachClientdeclares its raw stream class and multipart limits as class attributes, so custom client subclasses inherit streaming.cloudpathlib/env.py: environment-variable handling shared by the library.Content-Encoding, which made.gzobjects transcode on read.get_object, multipart,upload_from_file), a thread-safe Azure metadata cache, and a fix for the ADLS gen2 mock reporting aNonesize.docs/docs/streaming_io.mdand a new section in the caching notebook.Testing
USE_LIVE_CLOUD=1 CLOUDPATHLIB_TEST_RIGS=s3,gs,azure,azure_gen2 python -m pytest tests/test_cloud_io.py.Closes #9