diff --git a/CHANGELOG.md b/CHANGELOG.md index b0ced3c7..b0bcfb50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ Unreleased ---------- -- . +- Added a `content_settings` parameter to `open()` (write modes), `pipe_file()` and `put_file()` to set blob content settings (content type, content disposition, cache control, ...) on write. Accepts a `dict` (recommended) or an `azure.storage.blob.ContentSettings` instance. [#554](https://github.com/fsspec/adlfs/pull/554) 2026.5.0 -------- diff --git a/adlfs/spec.py b/adlfs/spec.py index 31a26121..fba2e761 100644 --- a/adlfs/spec.py +++ b/adlfs/spec.py @@ -31,6 +31,7 @@ BlobProperties, BlobSasPermissions, BlobType, + ContentSettings, generate_blob_sas, ) from azure.storage.blob.aio import BlobPrefix @@ -163,6 +164,19 @@ def _normalize_etag_quotes(etag: str) -> Optional[str]: return f'"{etag.strip(double_quote)}"' +def _as_content_settings( + content_settings: Optional[Union[ContentSettings, dict]], +) -> Optional[ContentSettings]: + """Coerce ``content_settings`` given as a dict into a ``ContentSettings``. + + Accepts either a plain ``dict`` (recommended — keeps the Azure SDK out of the + caller's imports) or a ``ContentSettings`` instance. + """ + if content_settings is None or isinstance(content_settings, ContentSettings): + return content_settings + return ContentSettings(**content_settings) + + class AzureBlobFileSystem(AsyncFileSystem): """ Access Azure Datalake Gen2 and Azure Storage if it were a file system using Multiprotocol Access @@ -1505,6 +1519,7 @@ async def _pipe_file( """Set the bytes of given file""" if kwargs.pop("mode", "") == "create": overwrite = False + content_settings = _as_content_settings(kwargs.pop("content_settings", None)) container_name, path, _ = self.split_path(path) async with self.service_client.get_blob_client( container=container_name, blob=path @@ -1514,6 +1529,7 @@ async def _pipe_file( data=value, overwrite=overwrite, metadata={"is_directory": "false"}, + content_settings=content_settings, max_concurrency=max_concurrency or self.max_concurrency, **self._timeout_kwargs, **kwargs, @@ -1744,6 +1760,7 @@ async def _put_file( if kwargs.pop("mode", "") == "create": overwrite = False + content_settings = _as_content_settings(kwargs.pop("content_settings", None)) container_name, path, _ = self.split_path(rpath, delimiter=delimiter) if os.path.isdir(lpath): @@ -1758,6 +1775,7 @@ async def _put_file( f1, overwrite=overwrite, metadata={"is_directory": "false"}, + content_settings=content_settings, raw_response_hook=make_callback( "upload_stream_current", callback ), @@ -1899,6 +1917,7 @@ def _open( cache_type="readahead", metadata=None, version_id: Optional[str] = None, + content_settings: Optional[Union[ContentSettings, dict]] = None, **kwargs, ): """Open a file on the datalake, or a block blob @@ -1927,6 +1946,11 @@ def _open( version_id: str Explicit version of the blob to open. This requires that the abfs filesystem is versioning aware and blob versioning is enabled on the releveant container. + + content_settings: dict + Optional content settings applied to the blob when writing, e.g. + ``{"content_type": "application/pdf", "content_disposition": "..."}``. + A ``ContentSettings`` instance is also accepted. """ logger.debug(f"_open: {path}") if block_size is None: @@ -1946,6 +1970,7 @@ def _open( cache_type=cache_type, metadata=metadata, version_id=version_id, + content_settings=content_settings, **kwargs, ) @@ -1966,6 +1991,7 @@ def __init__( cache_options: dict = {}, metadata=None, version_id: Optional[str] = None, + content_settings: Optional[Union[ContentSettings, dict]] = None, **kwargs, ): """ @@ -2003,6 +2029,11 @@ def __init__( attribute is populated with the version created by the upload when the filesystem has ``version_aware=True``. + content_settings: dict + Optional content settings applied to the blob when writing, e.g. + ``{"content_type": "application/pdf", "content_disposition": "..."}``. + A ``ContentSettings`` instance is also accepted. + kwargs: dict Passed to AbstractBufferedFile """ @@ -2073,6 +2104,7 @@ def __init__( else: self._metadata = metadata or {"is_directory": "false"} + self._content_settings = _as_content_settings(content_settings) self.buffer = io.BytesIO() self.offset = None self.forced = False @@ -2209,7 +2241,10 @@ async def _async_initiate_upload(self, **kwargs): if self.mode == "ab": if not await self.fs._exists(self.path): async with self.container_client.get_blob_client(blob=self.blob) as bc: - await bc.create_append_blob(metadata=self.metadata) + await bc.create_append_blob( + metadata=self.metadata, + content_settings=self._content_settings, + ) _initiate_upload = sync_wrapper(_async_initiate_upload) @@ -2269,7 +2304,10 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs): blob=self.blob ) as bc: response = await bc.commit_block_list( - block_list=block_list, metadata=self.metadata, **commit_kw + block_list=block_list, + metadata=self.metadata, + content_settings=self._content_settings, + **commit_kw, ) if self.fs.version_aware: self.version_id = response.get("version_id") @@ -2287,6 +2325,7 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs): response = await bc.upload_blob( data=data, metadata=self.metadata, + content_settings=self._content_settings, overwrite=(self.mode == "wb"), ) if self.fs.version_aware: @@ -2301,6 +2340,7 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs): response = await bc.commit_block_list( block_list=block_list, metadata=self.metadata, + content_settings=self._content_settings, **commit_kw, ) if self.fs.version_aware: diff --git a/adlfs/tests/test_spec.py b/adlfs/tests/test_spec.py index d2a63dee..d98ab94c 100644 --- a/adlfs/tests/test_spec.py +++ b/adlfs/tests/test_spec.py @@ -1413,6 +1413,101 @@ def test_metadata_write(storage): fs.rmdir("test-metadata-write") +CONTENT_SETTINGS = { + "content_type": "application/pdf", + "content_disposition": 'attachment; filename="f.pdf"', + "cache_control": "max-age=3600", +} + + +def test_content_settings_open(storage): + fs = AzureBlobFileSystem( + account_name=storage.account_name, connection_string=CONN_STR + ) + fs.mkdir("test-cs-open") + with fs.open("test-cs-open/file.pdf", "wb", content_settings=CONTENT_SETTINGS) as f: + f.write(b"0123456789") + cs = fs.info("test-cs-open/file.pdf")["content_settings"] + assert cs["content_type"] == "application/pdf" + assert cs["content_disposition"] == 'attachment; filename="f.pdf"' + assert cs["cache_control"] == "max-age=3600" + + # empty streaming write still applies content settings + with fs.open( + "test-cs-open/empty.pdf", "wb", content_settings=CONTENT_SETTINGS + ) as f: + f.write(b"") + assert ( + fs.info("test-cs-open/empty.pdf")["content_settings"]["content_type"] + == "application/pdf" + ) + fs.rmdir("test-cs-open") + + +def test_content_settings_append(storage): + fs = AzureBlobFileSystem( + account_name=storage.account_name, connection_string=CONN_STR + ) + fs.mkdir("test-cs-append") + with fs.open( + "test-cs-append/file.pdf", "ab", content_settings=CONTENT_SETTINGS + ) as f: + f.write(b"0123456789") + cs = fs.info("test-cs-append/file.pdf")["content_settings"] + assert cs["content_type"] == "application/pdf" + assert cs["content_disposition"] == 'attachment; filename="f.pdf"' + fs.rmdir("test-cs-append") + + +def test_content_settings_pipe_file(storage): + fs = AzureBlobFileSystem( + account_name=storage.account_name, connection_string=CONN_STR + ) + fs.mkdir("test-cs-pipe") + fs.pipe_file( + "test-cs-pipe/piped.pdf", b"0123456789", content_settings=CONTENT_SETTINGS + ) + info = fs.info("test-cs-pipe/piped.pdf") + assert info["content_settings"]["content_disposition"] == ( + 'attachment; filename="f.pdf"' + ) + assert fs.cat_file("test-cs-pipe/piped.pdf") == b"0123456789" + fs.rmdir("test-cs-pipe") + + +def test_content_settings_put_file(storage, tmp_path): + fs = AzureBlobFileSystem( + account_name=storage.account_name, connection_string=CONN_STR + ) + fs.mkdir("test-cs-put") + local = tmp_path / "f.pdf" + local.write_bytes(b"0123456789") + fs.put_file(str(local), "test-cs-put/put.pdf", content_settings=CONTENT_SETTINGS) + info = fs.info("test-cs-put/put.pdf") + assert info["content_settings"]["content_type"] == "application/pdf" + assert fs.cat_file("test-cs-put/put.pdf") == b"0123456789" + fs.rmdir("test-cs-put") + + +def test_content_settings_accepts_object(storage): + # A ContentSettings instance is accepted for compatibility (dict is the + # documented form). + from azure.storage.blob import ContentSettings + + fs = AzureBlobFileSystem( + account_name=storage.account_name, connection_string=CONN_STR + ) + fs.mkdir("test-cs-object") + settings = ContentSettings(content_type="application/pdf") + with fs.open("test-cs-object/obj.pdf", "wb", content_settings=settings) as f: + f.write(b"x") + assert ( + fs.info("test-cs-object/obj.pdf")["content_settings"]["content_type"] + == "application/pdf" + ) + fs.rmdir("test-cs-object") + + def test_put_file(storage, tmp_path): fs = AzureBlobFileSystem( account_name=storage.account_name, connection_string=CONN_STR @@ -2012,6 +2107,7 @@ async def test_pipe_file_timeout(storage, mocker): upload_blob.assert_called_once_with( data=b"data", metadata={"is_directory": "false"}, + content_settings=None, overwrite=True, max_concurrency=fs.max_concurrency, timeout=11, @@ -2040,6 +2136,7 @@ async def test_put_file_timeout(storage, mocker, tmp_path): upload_blob.assert_called_once_with( mocker.ANY, metadata={"is_directory": "false"}, + content_settings=None, overwrite=True, raw_response_hook=None, max_concurrency=fs.max_concurrency,